add and use a deduplicating source (#185)

* feat: add and use a deduplicating source

* chore: add log entry when endpoint was deduplicated

main.go

@@ -86,7 +86,7 @@ func main() {
 		log.Fatal(err)
 	}
 
-	multiSource := source.NewMultiSource(sources)
+	endpointsSource := source.NewDedupSource(source.NewMultiSource(sources))
 
 	var p provider.Provider
 	switch cfg.Provider {
@@ -121,7 +121,7 @@ func main() {
 	}
 
 	ctrl := controller.Controller{
-		Source:   multiSource,
+		Source:   endpointsSource,
 		Registry: r,
 		Policy:   policy,
 		Interval: cfg.Interval,

source/dedup_source.go

@@ -0,0 +1,58 @@
+/*
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package source
+
+import (
+	log "github.com/Sirupsen/logrus"
+
+	"github.com/kubernetes-incubator/external-dns/endpoint"
+)
+
+// dedupSource is a Source that removes duplicate endpoints from its wrapped source.
+type dedupSource struct {
+	source Source
+}
+
+// NewDedupSource creates a new dedupSource wrapping the provided Source.
+func NewDedupSource(source Source) Source {
+	return &dedupSource{source: source}
+}
+
+// Endpoints collects endpoints from its wrapped source and returns them without duplicates.
+func (ms *dedupSource) Endpoints() ([]*endpoint.Endpoint, error) {
+	result := []*endpoint.Endpoint{}
+	collected := map[string]bool{}
+
+	endpoints, err := ms.source.Endpoints()
+	if err != nil {
+		return nil, err
+	}
+
+	for _, ep := range endpoints {
+		identifier := ep.DNSName + " / " + ep.Target
+
+		if _, ok := collected[identifier]; ok {
+			log.Debugf("Removing duplicate endpoint %s", ep)
+			continue
+		}
+
+		collected[identifier] = true
+		result = append(result, ep)
+	}
+
+	return result, nil
+}

source/dedup_source_test.go

@@ -0,0 +1,105 @@
+/*
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package source
+
+import (
+	"testing"
+
+	"github.com/kubernetes-incubator/external-dns/endpoint"
+)
+
+// Validates that dedupSource is a Source
+var _ Source = &dedupSource{}
+
+func TestDedup(t *testing.T) {
+	t.Run("Endpoints", testDedupEndpoints)
+}
+
+// testDedupEndpoints tests that duplicates from the wrapped source are removed.
+func testDedupEndpoints(t *testing.T) {
+	for _, tc := range []struct {
+		title     string
+		endpoints []*endpoint.Endpoint
+		expected  []*endpoint.Endpoint
+	}{
+		{
+			"one endpoint returns one endpoint",
+			[]*endpoint.Endpoint{
+				{DNSName: "foo.example.org", Target: "1.2.3.4"},
+			},
+			[]*endpoint.Endpoint{
+				{DNSName: "foo.example.org", Target: "1.2.3.4"},
+			},
+		},
+		{
+			"two different endpoints return two endpoints",
+			[]*endpoint.Endpoint{
+				{DNSName: "foo.example.org", Target: "1.2.3.4"},
+				{DNSName: "bar.example.org", Target: "4.5.6.7"},
+			},
+			[]*endpoint.Endpoint{
+				{DNSName: "foo.example.org", Target: "1.2.3.4"},
+				{DNSName: "bar.example.org", Target: "4.5.6.7"},
+			},
+		},
+		{
+			"two endpoints with same dnsname and different targets return two endpoints",
+			[]*endpoint.Endpoint{
+				{DNSName: "foo.example.org", Target: "1.2.3.4"},
+				{DNSName: "foo.example.org", Target: "4.5.6.7"},
+			},
+			[]*endpoint.Endpoint{
+				{DNSName: "foo.example.org", Target: "1.2.3.4"},
+				{DNSName: "foo.example.org", Target: "4.5.6.7"},
+			},
+		},
+		{
+			"two endpoints with different dnsname and same target return two endpoints",
+			[]*endpoint.Endpoint{
+				{DNSName: "foo.example.org", Target: "1.2.3.4"},
+				{DNSName: "bar.example.org", Target: "1.2.3.4"},
+			},
+			[]*endpoint.Endpoint{
+				{DNSName: "foo.example.org", Target: "1.2.3.4"},
+				{DNSName: "bar.example.org", Target: "1.2.3.4"},
+			},
+		},
+		{
+			"two endpoints with same dnsname and same target return one endpoint",
+			[]*endpoint.Endpoint{
+				{DNSName: "foo.example.org", Target: "1.2.3.4"},
+				{DNSName: "foo.example.org", Target: "1.2.3.4"},
+			},
+			[]*endpoint.Endpoint{
+				{DNSName: "foo.example.org", Target: "1.2.3.4"},
+			},
+		},
+	} {
+		t.Run(tc.title, func(t *testing.T) {
+			// Create our object under test and get the endpoints.
+			source := NewDedupSource(NewMockSource(tc.endpoints))
+
+			endpoints, err := source.Endpoints()
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			// Validate returned endpoints against desired endpoints.
+			validateEndpoints(t, endpoints, tc.expected)
+		})
+	}
+}