Merge pull request #84 from linki/annotations

make mate play well with other ext dns controllers

config.go

@@ -23,6 +23,7 @@ type mateConfig struct {
 	kubernetesServer         *url.URL
 	kubernetesFormat         string
 	kubernetesTrackNodePorts bool
+	kubernetesFilter         map[string]string
 
 	awsRecordGroupID string
 
@@ -32,7 +33,7 @@ type mateConfig struct {
 
 func newConfig(version string) *mateConfig {
 	kingpin.Version(version)
-	return &mateConfig{}
+	return &mateConfig{kubernetesFilter: map[string]string{}}
 }
 
 func (cfg *mateConfig) parseFlags() {
@@ -51,6 +52,7 @@ func (cfg *mateConfig) parseFlags() {
 	kingpin.Flag("kubernetes-server", "The address of the Kubernetes API server.").URLVar(&cfg.kubernetesServer)
 	kingpin.Flag("kubernetes-format", "Format of DNS entries, e.g. {{.Name}}-{{.Namespace}}.example.com").StringVar(&cfg.kubernetesFormat)
 	kingpin.Flag("kubernetes-track-node-ports", "When true, generates DNS entries for type=NodePort services").BoolVar(&cfg.kubernetesTrackNodePorts)
+	kingpin.Flag("kubernetes-filter", "A set of annotations that must match in order to process the object.").StringMapVar(&cfg.kubernetesFilter)
 
 	kingpin.Flag("aws-record-group-id", "Identifier to filter mate created records ").StringVar(&cfg.awsRecordGroupID)
 

main.go

@@ -76,6 +76,7 @@ func newProducer(cfg *mateConfig) (producers.Producer, error) {
 			Format:         cfg.kubernetesFormat,
 			APIServer:      cfg.kubernetesServer,
 			TrackNodePorts: cfg.kubernetesTrackNodePorts,
+			Filter:         cfg.kubernetesFilter,
 		}
 		return producers.NewKubernetesProducer(kubeConfig)
 	case "fake":

producers/ingress.go

@@ -20,6 +20,7 @@ import (
 type kubernetesIngressProducer struct {
 	client *k8s.Clientset
 	tmpl   *template.Template
+	filter map[string]string
 }
 
 func NewKubernetesIngress(cfg *KubernetesOptions) (*kubernetesIngressProducer, error) {
@@ -38,6 +39,7 @@ func NewKubernetesIngress(cfg *KubernetesOptions) (*kubernetesIngressProducer, e
 	return &kubernetesIngressProducer{
 		client: client,
 		tmpl:   tmpl,
+		filter: cfg.Filter,
 	}, nil
 }
 
@@ -50,7 +52,7 @@ func (a *kubernetesIngressProducer) Endpoints() ([]*pkg.Endpoint, error) {
 	endpoints := make([]*pkg.Endpoint, 0)
 
 	for _, ing := range allIngress.Items {
-		if err := validateIngress(ing); err != nil {
+		if err := validateIngress(ing, a.filter); err != nil {
 			log.Warnln(err)
 			continue
 		}
@@ -107,7 +109,7 @@ loop:
 
 				log.Printf("%s: %s/%s", event.Type, ing.Namespace, ing.Name)
 
-				if err := validateIngress(*ing); err != nil {
+				if err := validateIngress(*ing, a.filter); err != nil {
 					log.Warnln(err)
 					continue
 				}
@@ -125,7 +127,16 @@ loop:
 	}
 }
 
-func validateIngress(ing extensions.Ingress) error {
+func validateIngress(ing extensions.Ingress, filter map[string]string) error {
+	for key := range filter {
+		if ing.Annotations[key] != filter[key] {
+			return fmt.Errorf(
+				"[Ingress] Ingress '%s/%s' doesn't match filter for annotation %s: %s != %s",
+				ing.Namespace, ing.Name, key, filter[key], ing.Annotations[key],
+			)
+		}
+	}
+
 	switch {
 	case len(ing.Status.LoadBalancer.Ingress) == 0:
 		return fmt.Errorf(

producers/ingress_test.go

@@ -0,0 +1,42 @@
+package producers
+
+import (
+	"testing"
+
+	"k8s.io/client-go/pkg/api/v1"
+	extensions "k8s.io/client-go/pkg/apis/extensions/v1beta1"
+)
+
+func TestValidateIngress(t *testing.T) {
+	loadBalancer := v1.LoadBalancerStatus{
+		Ingress: []v1.LoadBalancerIngress{v1.LoadBalancerIngress{IP: "8.8.8.8"}},
+	}
+
+	emptyIngress := extensions.Ingress{}
+
+	validIngress := extensions.Ingress{
+		Status: extensions.IngressStatus{LoadBalancer: loadBalancer},
+	}
+
+	validMatchedIngress := extensions.Ingress{
+		ObjectMeta: v1.ObjectMeta{Annotations: map[string]string{"foo": "bar"}},
+		Status:     extensions.IngressStatus{LoadBalancer: loadBalancer},
+	}
+
+	for _, test := range []struct {
+		ingress extensions.Ingress
+		filter  map[string]string
+		isErr   bool
+	}{
+		{emptyIngress, map[string]string{}, false},
+		{validIngress, map[string]string{}, true},
+		{validIngress, map[string]string{"foo": "bar"}, false},
+		{validMatchedIngress, map[string]string{"foo": "bar"}, true},
+		{validMatchedIngress, map[string]string{"foo": "qux"}, false},
+	} {
+		result := validateIngress(test.ingress, test.filter)
+		if _, isErr := result.(error); isErr == test.isErr {
+			t.Errorf("validateIngress(%q, %q) => %q, want %t", test.ingress.Name, test.filter, result, test.isErr)
+		}
+	}
+}

producers/kubernetes.go

@@ -25,6 +25,7 @@ type KubernetesOptions struct {
 	APIServer      *url.URL
 	Format         string
 	TrackNodePorts bool
+	Filter         map[string]string
 }
 
 func NewKubernetesProducer(cfg *KubernetesOptions) (*kubernetesProducer, error) {

producers/service.go

@@ -20,6 +20,7 @@ import (
 type kubernetesServiceProducer struct {
 	client *k8s.Clientset
 	tmpl   *template.Template
+	filter map[string]string
 }
 
 func NewKubernetesService(cfg *KubernetesOptions) (*kubernetesServiceProducer, error) {
@@ -38,6 +39,7 @@ func NewKubernetesService(cfg *KubernetesOptions) (*kubernetesServiceProducer, e
 	return &kubernetesServiceProducer{
 		client: client,
 		tmpl:   tmpl,
+		filter: cfg.Filter,
 	}, nil
 }
 
@@ -50,7 +52,7 @@ func (a *kubernetesServiceProducer) Endpoints() ([]*pkg.Endpoint, error) {
 	endpoints := make([]*pkg.Endpoint, 0)
 
 	for _, svc := range allServices.Items {
-		if err := validateService(svc); err != nil {
+		if err := validateService(svc, a.filter); err != nil {
 			log.Warnln(err)
 			continue
 		}
@@ -112,7 +114,7 @@ loop:
 
 				log.Printf("%s: %s/%s", event.Type, svc.Namespace, svc.Name)
 
-				if err := validateService(*svc); err != nil {
+				if err := validateService(*svc, a.filter); err != nil {
 					log.Warnln(err)
 					continue
 				}
@@ -132,7 +134,16 @@ loop:
 	}
 }
 
-func validateService(svc api.Service) error {
+func validateService(svc api.Service, filter map[string]string) error {
+	for key := range filter {
+		if svc.Annotations[key] != filter[key] {
+			return fmt.Errorf(
+				"[Service] Service '%s/%s' doesn't match filter for annotation %s: %s != %s",
+				svc.Namespace, svc.Name, key, filter[key], svc.Annotations[key],
+			)
+		}
+	}
+
 	switch {
 	case len(svc.Status.LoadBalancer.Ingress) == 0:
 		return fmt.Errorf(

producers/service_test.go

@@ -0,0 +1,41 @@
+package producers
+
+import (
+	"testing"
+
+	"k8s.io/client-go/pkg/api/v1"
+)
+
+func TestValidateService(t *testing.T) {
+	loadBalancer := v1.LoadBalancerStatus{
+		Ingress: []v1.LoadBalancerIngress{v1.LoadBalancerIngress{IP: "8.8.8.8"}},
+	}
+
+	emptyService := v1.Service{}
+
+	validService := v1.Service{
+		Status: v1.ServiceStatus{LoadBalancer: loadBalancer},
+	}
+
+	validMatchedService := v1.Service{
+		ObjectMeta: v1.ObjectMeta{Annotations: map[string]string{"foo": "bar"}},
+		Status:     v1.ServiceStatus{LoadBalancer: loadBalancer},
+	}
+
+	for _, test := range []struct {
+		service v1.Service
+		filter  map[string]string
+		isErr   bool
+	}{
+		{emptyService, map[string]string{}, false},
+		{validService, map[string]string{}, true},
+		{validService, map[string]string{"foo": "bar"}, false},
+		{validMatchedService, map[string]string{"foo": "bar"}, true},
+		{validMatchedService, map[string]string{"foo": "qux"}, false},
+	} {
+		result := validateService(test.service, test.filter)
+		if _, isErr := result.(error); isErr == test.isErr {
+			t.Errorf("validateService(%q, %q) => %q, want %t", test.service.Name, test.filter, result, test.isErr)
+		}
+	}
+}