implemented annotationFilter for node source

source/node.go

@@ -25,6 +25,7 @@ import (
 
 	log "github.com/sirupsen/logrus"
 	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/wait"
 	kubeinformers "k8s.io/client-go/informers"
@@ -37,11 +38,12 @@ import (
 
 type nodeSource struct {
 	client           kubernetes.Interface
+	annotationFilter string
 	fqdnTemplate     *template.Template
 	nodeInformer     coreinformers.NodeInformer
 }
 
-func NewNodeSource(kubeClient kubernetes.Interface, fqdnTemplate string) (Source, error) {
+func NewNodeSource(kubeClient kubernetes.Interface, annotationFilter, fqdnTemplate string) (Source, error) {
 	var (
 		tmpl *template.Template
 		err  error
@@ -83,6 +85,7 @@ func NewNodeSource(kubeClient kubernetes.Interface, fqdnTemplate string) (Source
 
 	return &nodeSource{
 		client:           kubeClient,
+		annotationFilter: annotationFilter,
 		fqdnTemplate:     tmpl,
 		nodeInformer:     nodeInformer,
 	}, nil
@@ -95,6 +98,11 @@ func (ns *nodeSource) Endpoints() ([]*endpoint.Endpoint, error) {
 		return nil, err
 	}
 
+	nodes, err = ns.filterByAnnotations(nodes)
+	if err != nil {
+		return nil, err
+	}
+
 	endpoints := []*endpoint.Endpoint{}
 
 	// create endpoints for all nodes
@@ -149,3 +157,34 @@ func (ns *nodeSource) nodeAddress(node *v1.Node) (string, error) {
 
 	return "", fmt.Errorf("could not find node address for %s", node.Name)
 }
+
+// filterByAnnotations filters a list of nodes by a given annotation selector.
+func (ns *nodeSource) filterByAnnotations(nodes []*v1.Node) ([]*v1.Node, error) {
+	labelSelector, err := metav1.ParseToLabelSelector(ns.annotationFilter)
+	if err != nil {
+		return nil, err
+	}
+	selector, err := metav1.LabelSelectorAsSelector(labelSelector)
+	if err != nil {
+		return nil, err
+	}
+
+	// empty filter returns original list
+	if selector.Empty() {
+		return nodes, nil
+	}
+
+	filteredList := []*v1.Node{}
+
+	for _, node := range nodes {
+		// convert the node's annotations to an equivalent label selector
+		annotations := labels.Set(node.Annotations)
+
+		// include node if its annotations match the selector
+		if selector.Matches(annotations) {
+			filteredList = append(filteredList, node)
+		}
+	}
+
+	return filteredList, nil
+}

source/node_test.go

@@ -14,7 +14,7 @@ import (
 func TestNodeSource(t *testing.T) {
 	//suite.Run(t, new(ServiceSuite))
 	//t.Run("Interface", testServiceSourceImplementsSource)
-	//t.Run("NewNodeSource", testNodeSourceNewNodeSource)
+	t.Run("NewNodeSource", testNodeSourceNewNodeSource)
 	t.Run("Endpoints", testNodeSourceEndpoints)
 }
 
@@ -22,6 +22,7 @@ func TestNodeSource(t *testing.T) {
 func testNodeSourceNewNodeSource(t *testing.T) {
 	for _, ti := range []struct {
 		title            string
+		annotationFilter string
 		fqdnTemplate     string
 		expectError      bool
 	}{
@@ -39,10 +40,16 @@ func testNodeSourceNewNodeSource(t *testing.T) {
 			expectError:  false,
 			fqdnTemplate: "{{.Name}}-{{.Namespace}}.ext-dns.test.com",
 		},
+		{
+			title:            "non-empty annotation filter label",
+			expectError:      false,
+			annotationFilter: "kubernetes.io/ingress.class=nginx",
+		},
 	} {
 		t.Run(ti.title, func(t *testing.T) {
 			_, err := NewNodeSource(
 				fake.NewSimpleClientset(),
+				ti.annotationFilter,
 				ti.fqdnTemplate,
 			)
 
@@ -59,6 +66,7 @@ func testNodeSourceNewNodeSource(t *testing.T) {
 func testNodeSourceEndpoints(t *testing.T) {
 	for _, tc := range []struct {
 		title            string
+		annotationFilter string
 		fqdnTemplate     string
 		nodeName         string
 		nodeAddresses    []v1.NodeAddress
@@ -70,6 +78,7 @@ func testNodeSourceEndpoints(t *testing.T) {
 		{
 			"node with short hostname returns one endpoint",
 			"",
+			"",
 			"node1",
 			[]v1.NodeAddress{{v1.NodeExternalIP, "1.2.3.4"}},
 			map[string]string{},
@@ -82,6 +91,7 @@ func testNodeSourceEndpoints(t *testing.T) {
 		{
 			"node with fqdn returns one endpoint",
 			"",
+			"",
 			"node1.example.org",
 			[]v1.NodeAddress{{v1.NodeExternalIP, "1.2.3.4"}},
 			map[string]string{},
@@ -93,6 +103,7 @@ func testNodeSourceEndpoints(t *testing.T) {
 		},
 		{
 			"node with fqdn template returns endpoint with expanded hostname",
+			"",
 			"{{.Name}}.example.org",
 			"node1",
 			[]v1.NodeAddress{{v1.NodeExternalIP, "1.2.3.4"}},
@@ -105,6 +116,7 @@ func testNodeSourceEndpoints(t *testing.T) {
 		},
 		{
 			"node with fqdn and fqdn template returns one endpoint",
+			"",
 			"{{.Name}}.example.org",
 			"node1.example.org",
 			[]v1.NodeAddress{{v1.NodeExternalIP, "1.2.3.4"}},
@@ -118,6 +130,7 @@ func testNodeSourceEndpoints(t *testing.T) {
 		{
 			"node with both external and internal IP returns an endpoint with external IP",
 			"",
+			"",
 			"node1",
 			[]v1.NodeAddress{{v1.NodeExternalIP, "1.2.3.4"}, {v1.NodeInternalIP, "2.3.4.5"}},
 			map[string]string{},
@@ -130,6 +143,7 @@ func testNodeSourceEndpoints(t *testing.T) {
 		{
 			"node with only internal IP returns an endpoint with internal IP",
 			"",
+			"",
 			"node1",
 			[]v1.NodeAddress{{v1.NodeInternalIP, "2.3.4.5"}},
 			map[string]string{},
@@ -142,6 +156,7 @@ func testNodeSourceEndpoints(t *testing.T) {
 		{
 			"node with neither external nor internal IP returns no endpoints",
 			"",
+			"",
 			"node1",
 			[]v1.NodeAddress{},
 			map[string]string{},
@@ -149,6 +164,49 @@ func testNodeSourceEndpoints(t *testing.T) {
 			[]*endpoint.Endpoint{},
 			false,
 		},
+		{
+			"annotated node without annotation filter returns endpoint",
+			"",
+			"",
+			"node1",
+			[]v1.NodeAddress{{v1.NodeExternalIP, "1.2.3.4"}},
+			map[string]string{},
+			map[string]string{
+				"service.beta.kubernetes.io/external-traffic": "OnlyLocal",
+			},
+			[]*endpoint.Endpoint{
+				{RecordType: "A", DNSName: "node1", Targets: endpoint.Targets{"1.2.3.4"}},
+			},
+			false,
+		},
+		{
+			"annotated node with matching annotation filter returns endpoint",
+			"service.beta.kubernetes.io/external-traffic in (Global, OnlyLocal)",
+			"",
+			"node1",
+			[]v1.NodeAddress{{v1.NodeExternalIP, "1.2.3.4"}},
+			map[string]string{},
+			map[string]string{
+				"service.beta.kubernetes.io/external-traffic": "OnlyLocal",
+			},
+			[]*endpoint.Endpoint{
+				{RecordType: "A", DNSName: "node1", Targets: endpoint.Targets{"1.2.3.4"}},
+			},
+			false,
+		},
+		{
+			"annotated node with non-matching annotation filter returns endpoint",
+			"service.beta.kubernetes.io/external-traffic in (Global, OnlyLocal)",
+			"",
+			"node1",
+			[]v1.NodeAddress{{v1.NodeExternalIP, "1.2.3.4"}},
+			map[string]string{},
+			map[string]string{
+				"service.beta.kubernetes.io/external-traffic": "SomethingElse",
+			},
+			[]*endpoint.Endpoint{},
+			false,
+		},
 	} {
 		t.Run(tc.title, func(t *testing.T) {
 			// Create a Kubernetes testing client
@@ -171,6 +229,7 @@ func testNodeSourceEndpoints(t *testing.T) {
 			// Create our object under test and get the endpoints.
 			client, err := NewNodeSource(
 				kubernetes,
+				tc.annotationFilter,
 				tc.fqdnTemplate,
 			)
 			require.NoError(t, err)

source/store.go

@@ -158,7 +158,7 @@ func BuildWithConfig(source string, p ClientGenerator, cfg *Config) (Source, err
 		if err != nil {
 			return nil, err
 		}
-		return NewNodeSource(client, cfg.FQDNTemplate)
+		return NewNodeSource(client, cfg.AnnotationFilter, cfg.FQDNTemplate)
 	case "service":
 		client, err := p.KubeClient()
 		if err != nil {