Multi target plan ( Page Not Found ) (#404)

* Make suitableType() be Endpoint method

With this change it becomes possible to work with endpoint
of empty type in packages other than "provider". Also
it seems logical for a smart property getter without side effects
to be a method rather than a function in different package

* Make plan computation work correctly with multi-target domains

* fix drawing

* drop comments

* fix boilerplate header

* fix comment

* fix the bug with empty map

* rework registry to support random lables

*  serialize->serializeLabel function rename

* golint for err variable naming

* add additional test

* add tests for current case where one resource can generate multiple endpoints

* make labels have its own type, add serialization as a method

* add comment for exported error

* use greater rather than not equal zero

* update changelog

CHANGELOG.md

@@ -1,3 +1,9 @@
+  - Every record managed by External DNS is now mapped to a kubernetes resource (service/ingress) @ideahitme
+    - New field is stored in TXT DNS record which reflects which kubernetes resource has acquired the DNS name
+    - Target of DNS record is changed only if corresponding kubernetes resource target changes
+    - If kubernetes resource is deleted, then another resource may acquire DNS name
+    - "Flapping" target issue is resolved by providing a consistent and defined mechanism for choosing a target
+      
 ## v0.4.8 - 2017-11-22
 
   - Allow filtering by source annotation via `--annotation-filter` (#354) @khrisrichardson

endpoint/endpoint.go

@@ -22,8 +22,6 @@ import (
 )
 
 const (
-	// OwnerLabelKey is the name of the label that defines the owner of an Endpoint.
-	OwnerLabelKey = "owner"
 	// RecordTypeA is a RecordType enum value
 	RecordTypeA = "A"
 	// RecordTypeCNAME is a RecordType enum value
@@ -51,7 +49,7 @@ type Endpoint struct {
 	// TTL for the record
 	RecordTTL TTL
 	// Labels stores labels defined for the Endpoint
-	Labels map[string]string
+	Labels Labels
 }
 
 // NewEndpoint initialization method to be used to create an endpoint
@@ -65,20 +63,11 @@ func NewEndpointWithTTL(dnsName, target, recordType string, ttl TTL) *Endpoint {
 		DNSName:    strings.TrimSuffix(dnsName, "."),
 		Target:     strings.TrimSuffix(target, "."),
 		RecordType: recordType,
-		Labels:     map[string]string{},
+		Labels:     NewLabels(),
 		RecordTTL:  ttl,
 	}
 }
 
-// MergeLabels adds keys to labels if not defined for the endpoint
-func (e *Endpoint) MergeLabels(labels map[string]string) {
-	for k, v := range labels {
-		if e.Labels[k] == "" {
-			e.Labels[k] = v
-		}
-	}
-}
-
 func (e *Endpoint) String() string {
 	return fmt.Sprintf("%s %d IN %s %s", e.DNSName, e.RecordTTL, e.RecordType, e.Target)
 }

endpoint/endpoint_test.go

@@ -18,8 +18,6 @@ package endpoint
 
 import (
 	"testing"
-
-	"github.com/stretchr/testify/assert"
 )
 
 func TestNewEndpoint(t *testing.T) {
@@ -36,13 +34,3 @@ func TestNewEndpoint(t *testing.T) {
 		t.Error("endpoint is not initialized correctly")
 	}
 }
-
-func TestMergeLabels(t *testing.T) {
-	e := NewEndpoint("abc.com", "1.2.3.4", "A")
-	e.Labels = map[string]string{
-		"foo": "bar",
-		"baz": "qux",
-	}
-	e.MergeLabels(map[string]string{"baz": "baz", "new": "fox"})
-	assert.Equal(t, map[string]string{"foo": "bar", "baz": "qux", "new": "fox"}, e.Labels)
-}

endpoint/labels.go

@@ -0,0 +1,99 @@
+/*
+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 endpoint
+
+import (
+	"errors"
+	"fmt"
+	"sort"
+	"strings"
+)
+
+var (
+	// ErrInvalidHeritage is returned when heritage was not found, or different heritage is found
+	ErrInvalidHeritage = errors.New("heritage is unknown or not found")
+)
+
+const (
+	heritage = "external-dns"
+	// OwnerLabelKey is the name of the label that defines the owner of an Endpoint.
+	OwnerLabelKey = "owner"
+	// ResourceLabelKey is the name of the label that identifies k8s resource which wants to acquire the DNS name
+	ResourceLabelKey = "resource"
+)
+
+// Labels store metadata related to the endpoint
+// it is then stored in a persistent storage via serialization
+type Labels map[string]string
+
+// NewLabels returns empty Labels
+func NewLabels() Labels {
+	return map[string]string{}
+}
+
+// NewLabelsFromString constructs endpoints labels from a provided format string
+// if heritage set to another value is found then error is returned
+// no heritage automatically assumes is not owned by external-dns and returns invalidHeritage error
+func NewLabelsFromString(labelText string) (Labels, error) {
+	endpointLabels := map[string]string{}
+	labelText = strings.Trim(labelText, "\"") // drop quotes
+	tokens := strings.Split(labelText, ",")
+	foundExternalDNSHeritage := false
+	for _, token := range tokens {
+		if len(strings.Split(token, "=")) != 2 {
+			continue
+		}
+		key := strings.Split(token, "=")[0]
+		val := strings.Split(token, "=")[1]
+		if key == "heritage" && val != heritage {
+			return nil, ErrInvalidHeritage
+		}
+		if key == "heritage" {
+			foundExternalDNSHeritage = true
+			continue
+		}
+		if strings.HasPrefix(key, heritage) {
+			endpointLabels[strings.TrimPrefix(key, heritage+"/")] = val
+		}
+	}
+
+	if !foundExternalDNSHeritage {
+		return nil, ErrInvalidHeritage
+	}
+
+	return endpointLabels, nil
+}
+
+// Serialize transforms endpoints labels into a external-dns recognizable format string
+// withQuotes adds additional quotes
+func (l Labels) Serialize(withQuotes bool) string {
+	var tokens []string
+	tokens = append(tokens, fmt.Sprintf("heritage=%s", heritage))
+	var keys []string
+	for key := range l {
+		keys = append(keys, key)
+	}
+	sort.Strings(keys) // sort for consistency
+
+	for _, key := range keys {
+		tokens = append(tokens, fmt.Sprintf("%s/%s=%s", heritage, key, l[key]))
+	}
+	if withQuotes {
+		return fmt.Sprintf("\"%s\"", strings.Join(tokens, ","))
+	}
+	return strings.Join(tokens, ",")
+}

endpoint/labels_test.go

@@ -0,0 +1,92 @@
+/*
+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 endpoint
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/suite"
+)
+
+type LabelsSuite struct {
+	suite.Suite
+	foo                  Labels
+	fooAsText            string
+	fooAsTextWithQuotes  string
+	barText              string
+	barTextAsMap         Labels
+	noHeritageText       string
+	noHeritageAsMap      Labels
+	wrongHeritageText    string
+	multipleHeritageText string //considered invalid
+}
+
+func (suite *LabelsSuite) SetupTest() {
+	suite.foo = map[string]string{
+		"owner":    "foo-owner",
+		"resource": "foo-resource",
+	}
+	suite.fooAsText = "heritage=external-dns,external-dns/owner=foo-owner,external-dns/resource=foo-resource"
+	suite.fooAsTextWithQuotes = fmt.Sprintf(`"%s"`, suite.fooAsText)
+
+	suite.barTextAsMap = map[string]string{
+		"owner":    "bar-owner",
+		"resource": "bar-resource",
+		"new-key":  "bar-new-key",
+	}
+	suite.barText = "heritage=external-dns,,external-dns/owner=bar-owner,external-dns/resource=bar-resource,external-dns/new-key=bar-new-key,random=stuff,no-equal-sign,," //also has some random gibberish
+
+	suite.noHeritageText = "external-dns/owner=random-owner"
+	suite.wrongHeritageText = "heritage=mate,external-dns/owner=random-owner"
+	suite.multipleHeritageText = "heritage=mate,heritage=external-dns,external-dns/owner=random-owner"
+}
+
+func (suite *LabelsSuite) TestSerialize() {
+	suite.Equal(suite.fooAsText, suite.foo.Serialize(false), "should serializeLabel")
+	suite.Equal(suite.fooAsTextWithQuotes, suite.foo.Serialize(true), "should serializeLabel")
+}
+
+func (suite *LabelsSuite) TestDeserialize() {
+	foo, err := NewLabelsFromString(suite.fooAsText)
+	suite.NoError(err, "should succeed for valid label text")
+	suite.Equal(suite.foo, foo, "should reconstruct original label map")
+
+	foo, err = NewLabelsFromString(suite.fooAsTextWithQuotes)
+	suite.NoError(err, "should succeed for valid label text")
+	suite.Equal(suite.foo, foo, "should reconstruct original label map")
+
+	bar, err := NewLabelsFromString(suite.barText)
+	suite.NoError(err, "should succeed for valid label text")
+	suite.Equal(suite.barTextAsMap, bar, "should reconstruct original label map")
+
+	noHeritage, err := NewLabelsFromString(suite.noHeritageText)
+	suite.Equal(ErrInvalidHeritage, err, "should fail if no heritage is found")
+	suite.Nil(noHeritage, "should return nil")
+
+	wrongHeritage, err := NewLabelsFromString(suite.wrongHeritageText)
+	suite.Equal(ErrInvalidHeritage, err, "should fail if wrong heritage is found")
+	suite.Nil(wrongHeritage, "if error should return nil")
+
+	multipleHeritage, err := NewLabelsFromString(suite.multipleHeritageText)
+	suite.Equal(ErrInvalidHeritage, err, "should fail if multiple heritage is found")
+	suite.Nil(multipleHeritage, "if error should return nil")
+}
+
+func TestLabels(t *testing.T) {
+	suite.Run(t, new(LabelsSuite))
+}

internal/testutils/endpoint.go

@@ -16,8 +16,11 @@ limitations under the License.
 
 package testutils
 
-import "github.com/kubernetes-incubator/external-dns/endpoint"
-import "sort"
+import (
+	"sort"
+
+	"github.com/kubernetes-incubator/external-dns/endpoint"
+)
 
 /** test utility functions for endpoints verifications */
 
@@ -45,7 +48,8 @@ func (b byAllFields) Less(i, j int) bool {
 // considers example.org. and example.org DNSName/Target as different endpoints
 func SameEndpoint(a, b *endpoint.Endpoint) bool {
 	return a.DNSName == b.DNSName && a.Target == b.Target && a.RecordType == b.RecordType &&
-		a.Labels[endpoint.OwnerLabelKey] == b.Labels[endpoint.OwnerLabelKey] && a.RecordTTL == b.RecordTTL
+		a.Labels[endpoint.OwnerLabelKey] == b.Labels[endpoint.OwnerLabelKey] && a.RecordTTL == b.RecordTTL &&
+		a.Labels[endpoint.ResourceLabelKey] == b.Labels[endpoint.ResourceLabelKey]
 }
 
 // SameEndpoints compares two slices of endpoints regardless of order

plan/conflict.go

@@ -0,0 +1,70 @@
+/*
+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 plan
+
+import (
+	"sort"
+
+	"github.com/kubernetes-incubator/external-dns/endpoint"
+)
+
+// ConflictResolver is used to make a decision in case of two or more different kubernetes resources
+// are trying to acquire same DNS name
+type ConflictResolver interface {
+	ResolveCreate(candidates []*endpoint.Endpoint) *endpoint.Endpoint
+	ResolveUpdate(current *endpoint.Endpoint, candidates []*endpoint.Endpoint) *endpoint.Endpoint
+}
+
+// PerResource allows only one resource to own a given dns name
+type PerResource struct{}
+
+// ResolveCreate is invoked when dns name is not owned by any resource
+// ResolveCreate takes "minimal" (string comparison of Target) endpoint to acquire the DNS record
+func (s PerResource) ResolveCreate(candidates []*endpoint.Endpoint) *endpoint.Endpoint {
+	var min *endpoint.Endpoint
+	for _, ep := range candidates {
+		if min == nil || s.less(ep, min) {
+			min = ep
+		}
+	}
+	return min
+}
+
+// ResolveUpdate is invoked when dns name is already owned by "current" endpoint
+// ResolveUpdate uses "current" record as base and updates it accordingly with new version of same resource
+// if it doesn't exist then pick min
+func (s PerResource) ResolveUpdate(current *endpoint.Endpoint, candidates []*endpoint.Endpoint) *endpoint.Endpoint {
+	currentResource := current.Labels[endpoint.ResourceLabelKey] // resource which has already acquired the DNS
+	// TODO: sort candidates only needed because we can still have two endpoints from same resource here. We sort for consistency
+	// TODO: remove once single endpoint can have multiple targets
+	sort.SliceStable(candidates, func(i, j int) bool {
+		return s.less(candidates[i], candidates[j])
+	})
+	for _, ep := range candidates {
+		if ep.Labels[endpoint.ResourceLabelKey] == currentResource {
+			return ep
+		}
+	}
+	return s.ResolveCreate(candidates)
+}
+
+// less returns true if endpoint x is less than y
+func (s PerResource) less(x, y *endpoint.Endpoint) bool {
+	return x.Target < y.Target
+}
+
+// TODO: with cross-resource/cross-cluster setup alternative variations of ConflictResolver can be used

plan/conflict_test.go

@@ -0,0 +1,137 @@
+/*
+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 plan
+
+import (
+	"testing"
+
+	"github.com/kubernetes-incubator/external-dns/endpoint"
+	"github.com/stretchr/testify/suite"
+)
+
+var _ ConflictResolver = PerResource{}
+
+type ResolverSuite struct {
+	// resolvers
+	perResource PerResource
+	// endpoints
+	fooV1Cname          *endpoint.Endpoint
+	fooV2Cname          *endpoint.Endpoint
+	fooV2CnameDuplicate *endpoint.Endpoint
+	fooA5               *endpoint.Endpoint
+	bar127A             *endpoint.Endpoint
+	bar192A             *endpoint.Endpoint
+	bar127AAnother      *endpoint.Endpoint
+	legacyBar192A       *endpoint.Endpoint // record created in AWS now without resource label
+	suite.Suite
+}
+
+func (suite *ResolverSuite) SetupTest() {
+	suite.perResource = PerResource{}
+	// initialize endpoints used in tests
+	suite.fooV1Cname = &endpoint.Endpoint{
+		DNSName:    "foo",
+		Target:     "v1",
+		RecordType: "CNAME",
+		Labels: map[string]string{
+			endpoint.ResourceLabelKey: "ingress/default/foo-v1",
+		},
+	}
+	suite.fooV2Cname = &endpoint.Endpoint{
+		DNSName:    "foo",
+		Target:     "v2",
+		RecordType: "CNAME",
+		Labels: map[string]string{
+			endpoint.ResourceLabelKey: "ingress/default/foo-v2",
+		},
+	}
+	suite.fooV2CnameDuplicate = &endpoint.Endpoint{
+		DNSName:    "foo",
+		Target:     "v2",
+		RecordType: "CNAME",
+		Labels: map[string]string{
+			endpoint.ResourceLabelKey: "ingress/default/foo-v2-duplicate",
+		},
+	}
+	suite.fooA5 = &endpoint.Endpoint{
+		DNSName:    "foo",
+		Target:     "5.5.5.5",
+		RecordType: "A",
+		Labels: map[string]string{
+			endpoint.ResourceLabelKey: "ingress/default/foo-5",
+		},
+	}
+	suite.bar127A = &endpoint.Endpoint{
+		DNSName:    "bar",
+		Target:     "127.0.0.1",
+		RecordType: "A",
+		Labels: map[string]string{
+			endpoint.ResourceLabelKey: "ingress/default/bar-127",
+		},
+	}
+	suite.bar127AAnother = &endpoint.Endpoint{ //TODO: remove this once we move to multiple targets under same endpoint
+		DNSName:    "bar",
+		Target:     "8.8.8.8",
+		RecordType: "A",
+		Labels: map[string]string{
+			endpoint.ResourceLabelKey: "ingress/default/bar-127",
+		},
+	}
+	suite.bar192A = &endpoint.Endpoint{
+		DNSName:    "bar",
+		Target:     "192.168.0.1",
+		RecordType: "A",
+		Labels: map[string]string{
+			endpoint.ResourceLabelKey: "ingress/default/bar-192",
+		},
+	}
+	suite.legacyBar192A = &endpoint.Endpoint{
+		DNSName:    "bar",
+		Target:     "192.168.0.1",
+		RecordType: "A",
+	}
+}
+
+func (suite *ResolverSuite) TestStrictResolver() {
+	// test that perResource resolver picks min for create list
+	suite.Equal(suite.bar127A, suite.perResource.ResolveCreate([]*endpoint.Endpoint{suite.bar127A, suite.bar192A}), "should pick min one")
+	suite.Equal(suite.fooA5, suite.perResource.ResolveCreate([]*endpoint.Endpoint{suite.fooA5, suite.fooV1Cname}), "should pick min one")
+	suite.Equal(suite.fooV1Cname, suite.perResource.ResolveCreate([]*endpoint.Endpoint{suite.fooV2Cname, suite.fooV1Cname}), "should pick min one")
+
+	//test that perResource resolver preserves resource if it still exists
+	suite.Equal(suite.bar127A, suite.perResource.ResolveUpdate(suite.bar127A, []*endpoint.Endpoint{suite.bar127AAnother, suite.bar127A}), "should pick min for update when same resource endpoint occurs multiple times (remove after multiple-target support") // TODO:remove this test
+	suite.Equal(suite.bar127A, suite.perResource.ResolveUpdate(suite.bar127A, []*endpoint.Endpoint{suite.bar192A, suite.bar127A}), "should pick existing resource")
+	suite.Equal(suite.fooV2Cname, suite.perResource.ResolveUpdate(suite.fooV2Cname, []*endpoint.Endpoint{suite.fooV2Cname, suite.fooV2CnameDuplicate}), "should pick existing resource even if targets are same")
+	suite.Equal(suite.fooA5, suite.perResource.ResolveUpdate(suite.fooV1Cname, []*endpoint.Endpoint{suite.fooA5, suite.fooV2Cname}), "should pick new if resource was deleted")
+	// should actually get the updated record (note ttl is different)
+	newFooV1Cname := &endpoint.Endpoint{
+		DNSName:    suite.fooV1Cname.DNSName,
+		Target:     suite.fooV1Cname.Target,
+		Labels:     suite.fooV1Cname.Labels,
+		RecordType: suite.fooV1Cname.RecordType,
+		RecordTTL:  suite.fooV1Cname.RecordTTL + 1, // ttl is different
+	}
+	suite.Equal(newFooV1Cname, suite.perResource.ResolveUpdate(suite.fooV1Cname, []*endpoint.Endpoint{suite.fooA5, suite.fooV2Cname, newFooV1Cname}), "should actually pick same resource with updates")
+
+	// legacy record's resource value will not match any candidates resource label
+	// therefore pick minimum again
+	suite.Equal(suite.bar127A, suite.perResource.ResolveUpdate(suite.legacyBar192A, []*endpoint.Endpoint{suite.bar127A, suite.bar192A}), " legacy record's resource value will not match, should pick minimum")
+}
+
+func TestConflictResolver(t *testing.T) {
+	suite.Run(t, new(ResolverSuite))
+}

plan/plan.go

@@ -18,7 +18,6 @@ package plan
 
 import (
 	"github.com/kubernetes-incubator/external-dns/endpoint"
-	log "github.com/sirupsen/logrus"
 )
 
 // Plan can convert a list of desired and current records to a series of create,
@@ -47,55 +46,102 @@ type Changes struct {
 	Delete []*endpoint.Endpoint
 }
 
+// planTable is a supplementary struct for Plan
+// each row correspond to a dnsName -> (current record + all desired records)
+/*
+planTable: (-> = target)
+--------------------------------------------------------
+DNSName | Current record | Desired Records             |
+--------------------------------------------------------
+foo.com | -> 1.1.1.1     | [->1.1.1.1, ->elb.com]      |  = no action
+--------------------------------------------------------
+bar.com |                | [->191.1.1.1, ->190.1.1.1]  |  = create (bar.com -> 190.1.1.1)
+--------------------------------------------------------
+"=", i.e. result of calculation relies on supplied ConflictResolver
+*/
+type planTable struct {
+	rows     map[string]*planTableRow
+	resolver ConflictResolver
+}
+
+func newPlanTable() planTable { //TODO: make resolver configurable
+	return planTable{map[string]*planTableRow{}, PerResource{}}
+}
+
+// planTableRow
+// current corresponds to the record currently occupying dns name on the dns provider
+// candidates corresponds to the list of records which would like to have this dnsName
+type planTableRow struct {
+	current    *endpoint.Endpoint
+	candidates []*endpoint.Endpoint
+}
+
+func (t planTable) addCurrent(e *endpoint.Endpoint) {
+	if _, ok := t.rows[e.DNSName]; !ok {
+		t.rows[e.DNSName] = &planTableRow{}
+	}
+	t.rows[e.DNSName].current = e
+}
+
+func (t planTable) addCandidate(e *endpoint.Endpoint) {
+	if _, ok := t.rows[e.DNSName]; !ok {
+		t.rows[e.DNSName] = &planTableRow{}
+	}
+	t.rows[e.DNSName].candidates = append(t.rows[e.DNSName].candidates, e)
+}
+
+// TODO: allows record type change, which might not be supported by all dns providers
+func (t planTable) getUpdates() (updateNew []*endpoint.Endpoint, updateOld []*endpoint.Endpoint) {
+	for _, row := range t.rows {
+		if row.current != nil && len(row.candidates) > 0 { //dns name is taken
+			update := t.resolver.ResolveUpdate(row.current, row.candidates)
+			// compare "update" to "current" to figure out if actual update is required
+			if shouldUpdateTTL(update, row.current) || targetChanged(update, row.current) {
+				inheritOwner(row.current, update)
+				updateNew = append(updateNew, update)
+				updateOld = append(updateOld, row.current)
+			}
+			continue
+		}
+	}
+	return
+}
+
+func (t planTable) getCreates() (createList []*endpoint.Endpoint) {
+	for _, row := range t.rows {
+		if row.current == nil { //dns name not taken
+			createList = append(createList, t.resolver.ResolveCreate(row.candidates))
+		}
+	}
+	return
+}
+
+func (t planTable) getDeletes() (deleteList []*endpoint.Endpoint) {
+	for _, row := range t.rows {
+		if row.current != nil && len(row.candidates) == 0 {
+			deleteList = append(deleteList, row.current)
+		}
+	}
+	return
+}
+
 // Calculate computes the actions needed to move current state towards desired
 // state. It then passes those changes to the current policy for further
 // processing. It returns a copy of Plan with the changes populated.
 func (p *Plan) Calculate() *Plan {
-	changes := &Changes{}
+	t := newPlanTable()
 
-	// Ensure all desired records exist. For each desired record make sure it's
-	// either created or updated.
-	for _, desired := range p.Desired {
-		// Get the matching current record if it exists.
-		current, exists := recordExists(desired, p.Current)
-
-		// If there's no current record create desired record.
-		if !exists {
-			changes.Create = append(changes.Create, desired)
-			continue
-		}
-
-		targetChanged := targetChanged(desired, current)
-		shouldUpdateTTL := shouldUpdateTTL(desired, current)
-
-		if !targetChanged && !shouldUpdateTTL {
-			log.Debugf("Skipping endpoint %v because nothing has changed", desired)
-			continue
-		}
-
-		changes.UpdateOld = append(changes.UpdateOld, current)
-		desired.MergeLabels(current.Labels) // inherit the labels from the dns provider, including Owner ID
-
-		if targetChanged {
-			desired.RecordType = current.RecordType // inherit the type from the dns provider
-		}
-
-		if !shouldUpdateTTL {
-			desired.RecordTTL = current.RecordTTL
-		}
-
-		changes.UpdateNew = append(changes.UpdateNew, desired)
-	}
-
-	// Ensure all undesired records are removed. Each current record that cannot
-	// be found in the list of desired records is removed.
 	for _, current := range p.Current {
-		if _, exists := recordExists(current, p.Desired); !exists {
-			changes.Delete = append(changes.Delete, current)
-		}
+		t.addCurrent(current)
+	}
+	for _, desired := range p.Desired {
+		t.addCandidate(desired)
 	}
 
-	// Apply policies to list of changes.
+	changes := &Changes{}
+	changes.Create = t.getCreates()
+	changes.Delete = t.getDeletes()
+	changes.UpdateNew, changes.UpdateOld = t.getUpdates()
 	for _, pol := range p.Policies {
 		changes = pol.Apply(changes)
 	}
@@ -109,6 +155,16 @@ func (p *Plan) Calculate() *Plan {
 	return plan
 }
 
+func inheritOwner(from, to *endpoint.Endpoint) {
+	if to.Labels == nil {
+		to.Labels = map[string]string{}
+	}
+	if from.Labels == nil {
+		from.Labels = map[string]string{}
+	}
+	to.Labels[endpoint.OwnerLabelKey] = from.Labels[endpoint.OwnerLabelKey]
+}
+
 func targetChanged(desired, current *endpoint.Endpoint) bool {
 	return desired.Target != current.Target
 }
@@ -119,14 +175,3 @@ func shouldUpdateTTL(desired, current *endpoint.Endpoint) bool {
 	}
 	return desired.RecordTTL != current.RecordTTL
 }
-
-// recordExists checks whether a record can be found in a list of records.
-func recordExists(needle *endpoint.Endpoint, haystack []*endpoint.Endpoint) (*endpoint.Endpoint, bool) {
-	for _, record := range haystack {
-		if record.DNSName == needle.DNSName {
-			return record, true
-		}
-	}
-
-	return nil, false
-}

plan/plan_test.go

@@ -17,173 +17,343 @@ limitations under the License.
 package plan
 
 import (
-	"fmt"
 	"testing"
 
 	"github.com/kubernetes-incubator/external-dns/endpoint"
 	"github.com/kubernetes-incubator/external-dns/internal/testutils"
+	"github.com/stretchr/testify/suite"
 )
 
-// TestCalculate tests that a plan can calculate actions to move a list of
-// current records to a list of desired records.
-func TestCalculate(t *testing.T) {
-	// empty list of records
-	empty := []*endpoint.Endpoint{}
-	// a simple entry
-	fooV1 := []*endpoint.Endpoint{endpoint.NewEndpoint("foo", "v1", endpoint.RecordTypeCNAME)}
-	// the same entry but with different target
-	fooV2 := []*endpoint.Endpoint{endpoint.NewEndpoint("foo", "v2", endpoint.RecordTypeCNAME)}
-	// another simple entry
-	bar := []*endpoint.Endpoint{endpoint.NewEndpoint("bar", "v1", endpoint.RecordTypeCNAME)}
+type PlanTestSuite struct {
+	suite.Suite
+	fooV1Cname             *endpoint.Endpoint
+	fooV2Cname             *endpoint.Endpoint
+	fooV2CnameNoLabel      *endpoint.Endpoint
+	fooV3CnameSameResource *endpoint.Endpoint
+	fooA5                  *endpoint.Endpoint
+	bar127A                *endpoint.Endpoint
+	bar127AWithTTL         *endpoint.Endpoint
+	bar192A                *endpoint.Endpoint
+}
 
-	// test case with labels
-	noLabels := []*endpoint.Endpoint{endpoint.NewEndpoint("foo", "v2", endpoint.RecordTypeCNAME)}
-	labeledV2 := []*endpoint.Endpoint{newEndpointWithOwner("foo", "v2", "123")}
-	labeledV1 := []*endpoint.Endpoint{newEndpointWithOwner("foo", "v1", "123")}
-
-	// test case with type inheritance
-	noType := []*endpoint.Endpoint{endpoint.NewEndpoint("foo", "v2", "")}
-	typedV2 := []*endpoint.Endpoint{endpoint.NewEndpoint("foo", "v2", endpoint.RecordTypeA)}
-	typedV1 := []*endpoint.Endpoint{endpoint.NewEndpoint("foo", "v1", endpoint.RecordTypeA)}
-
-	// test case with TTL
-	ttl := endpoint.TTL(300)
-	ttl2 := endpoint.TTL(50)
-	ttlV1 := []*endpoint.Endpoint{endpoint.NewEndpointWithTTL("foo", "v1", endpoint.RecordTypeCNAME, ttl)}
-	ttlV2 := []*endpoint.Endpoint{endpoint.NewEndpoint("foo", "v1", endpoint.RecordTypeCNAME)}
-	ttlV3 := []*endpoint.Endpoint{endpoint.NewEndpointWithTTL("foo", "v1", endpoint.RecordTypeCNAME, ttl)}
-	ttlV4 := []*endpoint.Endpoint{endpoint.NewEndpointWithTTL("foo", "v1", endpoint.RecordTypeCNAME, ttl2)}
-	ttlV5 := []*endpoint.Endpoint{endpoint.NewEndpoint("foo", "v2", endpoint.RecordTypeCNAME)}
-	ttlV6 := []*endpoint.Endpoint{endpoint.NewEndpointWithTTL("foo", "v2", endpoint.RecordTypeCNAME, ttl)}
-
-	for _, tc := range []struct {
-		policies                             []Policy
-		current, desired                     []*endpoint.Endpoint
-		create, updateOld, updateNew, delete []*endpoint.Endpoint
-	}{
-		// Nothing exists and nothing desired doesn't change anything.
-		{[]Policy{&SyncPolicy{}}, empty, empty, empty, empty, empty, empty},
-		// More desired than current creates the desired.
-		{[]Policy{&SyncPolicy{}}, empty, fooV1, fooV1, empty, empty, empty},
-		// Desired equals current doesn't change anything.
-		{[]Policy{&SyncPolicy{}}, fooV1, fooV1, empty, empty, empty, empty},
-		// Nothing is desired deletes the current.
-		{[]Policy{&SyncPolicy{}}, fooV1, empty, empty, empty, empty, fooV1},
-		// Current and desired match but Target is different triggers an update.
-		{[]Policy{&SyncPolicy{}}, fooV1, fooV2, empty, fooV1, fooV2, empty},
-		// Both exist but are different creates desired and deletes current.
-		{[]Policy{&SyncPolicy{}}, fooV1, bar, bar, empty, empty, fooV1},
-		// Nothing is desired but policy doesn't allow deletions.
-		{[]Policy{&UpsertOnlyPolicy{}}, fooV1, empty, empty, empty, empty, empty},
-		// Labels should be inherited
-		{[]Policy{&SyncPolicy{}}, labeledV1, noLabels, empty, labeledV1, labeledV2, empty},
-		// RecordType should be inherited
-		{[]Policy{&SyncPolicy{}}, typedV1, noType, empty, typedV1, typedV2, empty},
-		// If desired TTL is not configured, do not update
-		{[]Policy{&SyncPolicy{}}, ttlV1, ttlV2, empty, empty, empty, empty},
-		// If desired TTL is configured but is the same as current TTL, do not update
-		{[]Policy{&SyncPolicy{}}, ttlV1, ttlV3, empty, empty, empty, empty},
-		// If desired TTL is configured and is not the same as current TTL, need to update
-		{[]Policy{&SyncPolicy{}}, ttlV1, ttlV4, empty, ttlV1, ttlV4, empty},
-		// If target changed and desired TTL is not configured, do not update TTL
-		{[]Policy{&SyncPolicy{}}, ttlV1, ttlV5, empty, ttlV1, ttlV6, empty},
-	} {
-		// setup plan
-		plan := &Plan{
-			Policies: tc.policies,
-			Current:  tc.current,
-			Desired:  tc.desired,
-		}
-		// calculate actions
-		plan = plan.Calculate()
-
-		// validate actions
-		validateEntries(t, plan.Changes.Create, tc.create)
-		validateEntries(t, plan.Changes.UpdateOld, tc.updateOld)
-		validateEntries(t, plan.Changes.UpdateNew, tc.updateNew)
-		validateEntries(t, plan.Changes.Delete, tc.delete)
+func (suite *PlanTestSuite) SetupTest() {
+	suite.fooV1Cname = &endpoint.Endpoint{
+		DNSName:    "foo",
+		Target:     "v1",
+		RecordType: "CNAME",
+		Labels: map[string]string{
+			endpoint.ResourceLabelKey: "ingress/default/foo-v1",
+			endpoint.OwnerLabelKey:    "pwner",
+		},
+	}
+	// same resource as fooV1Cname, but target is different. It will never be picked because its target lexicographically bigger than "v1"
+	suite.fooV3CnameSameResource = &endpoint.Endpoint{ // TODO: remove this once endpoint can support multiple targets
+		DNSName:    "foo",
+		Target:     "v3",
+		RecordType: "CNAME",
+		Labels: map[string]string{
+			endpoint.ResourceLabelKey: "ingress/default/foo-v1",
+			endpoint.OwnerLabelKey:    "pwner",
+		},
+	}
+	suite.fooV2Cname = &endpoint.Endpoint{
+		DNSName:    "foo",
+		Target:     "v2",
+		RecordType: "CNAME",
+		Labels: map[string]string{
+			endpoint.ResourceLabelKey: "ingress/default/foo-v2",
+		},
+	}
+	suite.fooV2CnameNoLabel = &endpoint.Endpoint{
+		DNSName:    "foo",
+		Target:     "v2",
+		RecordType: "CNAME",
+	}
+	suite.fooA5 = &endpoint.Endpoint{
+		DNSName:    "foo",
+		Target:     "5.5.5.5",
+		RecordType: "A",
+		Labels: map[string]string{
+			endpoint.ResourceLabelKey: "ingress/default/foo-5",
+		},
+	}
+	suite.bar127A = &endpoint.Endpoint{
+		DNSName:    "bar",
+		Target:     "127.0.0.1",
+		RecordType: "A",
+		Labels: map[string]string{
+			endpoint.ResourceLabelKey: "ingress/default/bar-127",
+		},
+	}
+	suite.bar127AWithTTL = &endpoint.Endpoint{
+		DNSName:    "bar",
+		Target:     "127.0.0.1",
+		RecordType: "A",
+		RecordTTL:  300,
+		Labels: map[string]string{
+			endpoint.ResourceLabelKey: "ingress/default/bar-127",
+		},
+	}
+	suite.bar192A = &endpoint.Endpoint{
+		DNSName:    "bar",
+		Target:     "192.168.0.1",
+		RecordType: "A",
+		Labels: map[string]string{
+			endpoint.ResourceLabelKey: "ingress/default/bar-192",
+		},
 	}
 }
 
-// BenchmarkCalculate benchmarks the Calculate method.
-func BenchmarkCalculate(b *testing.B) {
-	foo := endpoint.NewEndpoint("foo", "v1", "")
-	barV1 := endpoint.NewEndpoint("bar", "v1", "")
-	barV2 := endpoint.NewEndpoint("bar", "v2", "")
-	baz := endpoint.NewEndpoint("baz", "v1", "")
+func (suite *PlanTestSuite) TestSyncFirstRound() {
+	current := []*endpoint.Endpoint{}
+	desired := []*endpoint.Endpoint{suite.fooV1Cname, suite.fooV2Cname, suite.bar127A}
+	expectedCreate := []*endpoint.Endpoint{suite.fooV1Cname, suite.bar127A} //v1 is chosen because of resolver taking "min"
+	expectedUpdateOld := []*endpoint.Endpoint{}
+	expectedUpdateNew := []*endpoint.Endpoint{}
+	expectedDelete := []*endpoint.Endpoint{}
 
-	plan := &Plan{
-		Current: []*endpoint.Endpoint{foo, barV1},
-		Desired: []*endpoint.Endpoint{barV2, baz},
-	}
-
-	for i := 0; i < b.N; i++ {
-		plan.Calculate()
-	}
-}
-
-// ExamplePlan shows how plan can be used.
-func ExamplePlan() {
-	foo := endpoint.NewEndpoint("foo.example.com", "1.2.3.4", "")
-	barV1 := endpoint.NewEndpoint("bar.example.com", "8.8.8.8", "")
-	barV2 := endpoint.NewEndpoint("bar.example.com", "8.8.4.4", "")
-	baz := endpoint.NewEndpoint("baz.example.com", "6.6.6.6", "")
-
-	// Plan where
-	// * foo should be deleted
-	// * bar should be updated from v1 to v2
-	// * baz should be created
-	plan := &Plan{
+	p := &Plan{
 		Policies: []Policy{&SyncPolicy{}},
-		Current:  []*endpoint.Endpoint{foo, barV1},
-		Desired:  []*endpoint.Endpoint{barV2, baz},
+		Current:  current,
+		Desired:  desired,
 	}
 
-	// calculate actions
-	plan = plan.Calculate()
+	changes := p.Calculate().Changes
+	validateEntries(suite.T(), changes.Create, expectedCreate)
+	validateEntries(suite.T(), changes.UpdateNew, expectedUpdateNew)
+	validateEntries(suite.T(), changes.UpdateOld, expectedUpdateOld)
+	validateEntries(suite.T(), changes.Delete, expectedDelete)
+}
 
-	// print actions
-	fmt.Println("Create:")
-	for _, ep := range plan.Changes.Create {
-		fmt.Println(ep)
+func (suite *PlanTestSuite) TestSyncSecondRound() {
+	current := []*endpoint.Endpoint{suite.fooV1Cname}
+	desired := []*endpoint.Endpoint{suite.fooV2Cname, suite.fooV1Cname, suite.bar127A}
+	expectedCreate := []*endpoint.Endpoint{suite.bar127A}
+	expectedUpdateOld := []*endpoint.Endpoint{}
+	expectedUpdateNew := []*endpoint.Endpoint{}
+	expectedDelete := []*endpoint.Endpoint{}
+
+	p := &Plan{
+		Policies: []Policy{&SyncPolicy{}},
+		Current:  current,
+		Desired:  desired,
 	}
-	fmt.Println("UpdateOld:")
-	for _, ep := range plan.Changes.UpdateOld {
-		fmt.Println(ep)
+
+	changes := p.Calculate().Changes
+	validateEntries(suite.T(), changes.Create, expectedCreate)
+	validateEntries(suite.T(), changes.UpdateNew, expectedUpdateNew)
+	validateEntries(suite.T(), changes.UpdateOld, expectedUpdateOld)
+	validateEntries(suite.T(), changes.Delete, expectedDelete)
+}
+
+func (suite *PlanTestSuite) TestSyncSecondRoundMigration() {
+	current := []*endpoint.Endpoint{suite.fooV2CnameNoLabel}
+	desired := []*endpoint.Endpoint{suite.fooV2Cname, suite.fooV1Cname, suite.bar127A}
+	expectedCreate := []*endpoint.Endpoint{suite.bar127A}
+	expectedUpdateOld := []*endpoint.Endpoint{suite.fooV2CnameNoLabel}
+	expectedUpdateNew := []*endpoint.Endpoint{suite.fooV1Cname}
+	expectedDelete := []*endpoint.Endpoint{}
+
+	p := &Plan{
+		Policies: []Policy{&SyncPolicy{}},
+		Current:  current,
+		Desired:  desired,
 	}
-	fmt.Println("UpdateNew:")
-	for _, ep := range plan.Changes.UpdateNew {
-		fmt.Println(ep)
+
+	changes := p.Calculate().Changes
+	validateEntries(suite.T(), changes.Create, expectedCreate)
+	validateEntries(suite.T(), changes.UpdateNew, expectedUpdateNew)
+	validateEntries(suite.T(), changes.UpdateOld, expectedUpdateOld)
+	validateEntries(suite.T(), changes.Delete, expectedDelete)
+}
+
+func (suite *PlanTestSuite) TestSyncSecondRoundWithTTLChange() {
+	current := []*endpoint.Endpoint{suite.bar127A}
+	desired := []*endpoint.Endpoint{suite.bar127AWithTTL}
+	expectedCreate := []*endpoint.Endpoint{}
+	expectedUpdateOld := []*endpoint.Endpoint{suite.bar127A}
+	expectedUpdateNew := []*endpoint.Endpoint{suite.bar127AWithTTL}
+	expectedDelete := []*endpoint.Endpoint{}
+
+	p := &Plan{
+		Policies: []Policy{&SyncPolicy{}},
+		Current:  current,
+		Desired:  desired,
 	}
-	fmt.Println("Delete:")
-	for _, ep := range plan.Changes.Delete {
-		fmt.Println(ep)
+
+	changes := p.Calculate().Changes
+	validateEntries(suite.T(), changes.Create, expectedCreate)
+	validateEntries(suite.T(), changes.UpdateNew, expectedUpdateNew)
+	validateEntries(suite.T(), changes.UpdateOld, expectedUpdateOld)
+	validateEntries(suite.T(), changes.Delete, expectedDelete)
+}
+
+func (suite *PlanTestSuite) TestSyncSecondRoundWithOwnerInherited() {
+	current := []*endpoint.Endpoint{suite.fooV1Cname}
+	desired := []*endpoint.Endpoint{suite.fooV2Cname}
+
+	expectedCreate := []*endpoint.Endpoint{}
+	expectedUpdateOld := []*endpoint.Endpoint{suite.fooV1Cname}
+	expectedUpdateNew := []*endpoint.Endpoint{{
+		DNSName:    suite.fooV2Cname.DNSName,
+		Target:     suite.fooV2Cname.Target,
+		RecordType: suite.fooV2Cname.RecordType,
+		RecordTTL:  suite.fooV2Cname.RecordTTL,
+		Labels: map[string]string{
+			endpoint.ResourceLabelKey: suite.fooV2Cname.Labels[endpoint.ResourceLabelKey],
+			endpoint.OwnerLabelKey:    suite.fooV1Cname.Labels[endpoint.OwnerLabelKey],
+		},
+	}}
+	expectedDelete := []*endpoint.Endpoint{}
+
+	p := &Plan{
+		Policies: []Policy{&SyncPolicy{}},
+		Current:  current,
+		Desired:  desired,
 	}
-	// Create:
-	// &{baz.example.com 6.6.6.6 map[] }
-	// UpdateOld:
-	// &{bar.example.com 8.8.8.8 map[] }
-	// UpdateNew:
-	// &{bar.example.com 8.8.4.4 map[] }
-	// Delete:
-	// &{foo.example.com 1.2.3.4 map[] }
+
+	changes := p.Calculate().Changes
+	validateEntries(suite.T(), changes.Create, expectedCreate)
+	validateEntries(suite.T(), changes.UpdateNew, expectedUpdateNew)
+	validateEntries(suite.T(), changes.UpdateOld, expectedUpdateOld)
+	validateEntries(suite.T(), changes.Delete, expectedDelete)
+}
+
+func (suite *PlanTestSuite) TestIdempotency() {
+	current := []*endpoint.Endpoint{suite.fooV1Cname, suite.fooV2Cname}
+	desired := []*endpoint.Endpoint{suite.fooV1Cname, suite.fooV2Cname}
+	expectedCreate := []*endpoint.Endpoint{}
+	expectedUpdateOld := []*endpoint.Endpoint{}
+	expectedUpdateNew := []*endpoint.Endpoint{}
+	expectedDelete := []*endpoint.Endpoint{}
+
+	p := &Plan{
+		Policies: []Policy{&SyncPolicy{}},
+		Current:  current,
+		Desired:  desired,
+	}
+
+	changes := p.Calculate().Changes
+	validateEntries(suite.T(), changes.Create, expectedCreate)
+	validateEntries(suite.T(), changes.UpdateNew, expectedUpdateNew)
+	validateEntries(suite.T(), changes.UpdateOld, expectedUpdateOld)
+	validateEntries(suite.T(), changes.Delete, expectedDelete)
+}
+
+func (suite *PlanTestSuite) TestDifferentTypes() {
+	current := []*endpoint.Endpoint{suite.fooV1Cname}
+	desired := []*endpoint.Endpoint{suite.fooV2Cname, suite.fooA5}
+	expectedCreate := []*endpoint.Endpoint{}
+	expectedUpdateOld := []*endpoint.Endpoint{suite.fooV1Cname}
+	expectedUpdateNew := []*endpoint.Endpoint{suite.fooA5}
+	expectedDelete := []*endpoint.Endpoint{}
+
+	p := &Plan{
+		Policies: []Policy{&SyncPolicy{}},
+		Current:  current,
+		Desired:  desired,
+	}
+
+	changes := p.Calculate().Changes
+	validateEntries(suite.T(), changes.Create, expectedCreate)
+	validateEntries(suite.T(), changes.UpdateNew, expectedUpdateNew)
+	validateEntries(suite.T(), changes.UpdateOld, expectedUpdateOld)
+	validateEntries(suite.T(), changes.Delete, expectedDelete)
+}
+
+func (suite *PlanTestSuite) TestRemoveEndpoint() {
+	current := []*endpoint.Endpoint{suite.fooV1Cname, suite.bar192A}
+	desired := []*endpoint.Endpoint{suite.fooV1Cname}
+	expectedCreate := []*endpoint.Endpoint{}
+	expectedUpdateOld := []*endpoint.Endpoint{}
+	expectedUpdateNew := []*endpoint.Endpoint{}
+	expectedDelete := []*endpoint.Endpoint{suite.bar192A}
+
+	p := &Plan{
+		Policies: []Policy{&SyncPolicy{}},
+		Current:  current,
+		Desired:  desired,
+	}
+
+	changes := p.Calculate().Changes
+	validateEntries(suite.T(), changes.Create, expectedCreate)
+	validateEntries(suite.T(), changes.UpdateNew, expectedUpdateNew)
+	validateEntries(suite.T(), changes.UpdateOld, expectedUpdateOld)
+	validateEntries(suite.T(), changes.Delete, expectedDelete)
+}
+
+func (suite *PlanTestSuite) TestRemoveEndpointWithUpsert() {
+	current := []*endpoint.Endpoint{suite.fooV1Cname, suite.bar192A}
+	desired := []*endpoint.Endpoint{suite.fooV1Cname}
+	expectedCreate := []*endpoint.Endpoint{}
+	expectedUpdateOld := []*endpoint.Endpoint{}
+	expectedUpdateNew := []*endpoint.Endpoint{}
+	expectedDelete := []*endpoint.Endpoint{}
+
+	p := &Plan{
+		Policies: []Policy{&UpsertOnlyPolicy{}},
+		Current:  current,
+		Desired:  desired,
+	}
+
+	changes := p.Calculate().Changes
+	validateEntries(suite.T(), changes.Create, expectedCreate)
+	validateEntries(suite.T(), changes.UpdateNew, expectedUpdateNew)
+	validateEntries(suite.T(), changes.UpdateOld, expectedUpdateOld)
+	validateEntries(suite.T(), changes.Delete, expectedDelete)
+}
+
+//TODO: remove once multiple-target per endpoint is supported
+func (suite *PlanTestSuite) TestDuplicatedEndpointsForSameResourceReplace() {
+	current := []*endpoint.Endpoint{suite.fooV3CnameSameResource, suite.bar192A}
+	desired := []*endpoint.Endpoint{suite.fooV1Cname, suite.fooV3CnameSameResource}
+	expectedCreate := []*endpoint.Endpoint{}
+	expectedUpdateOld := []*endpoint.Endpoint{suite.fooV3CnameSameResource}
+	expectedUpdateNew := []*endpoint.Endpoint{suite.fooV1Cname}
+	expectedDelete := []*endpoint.Endpoint{suite.bar192A}
+
+	p := &Plan{
+		Policies: []Policy{&SyncPolicy{}},
+		Current:  current,
+		Desired:  desired,
+	}
+
+	changes := p.Calculate().Changes
+	validateEntries(suite.T(), changes.Create, expectedCreate)
+	validateEntries(suite.T(), changes.UpdateNew, expectedUpdateNew)
+	validateEntries(suite.T(), changes.UpdateOld, expectedUpdateOld)
+	validateEntries(suite.T(), changes.Delete, expectedDelete)
+}
+
+//TODO: remove once multiple-target per endpoint is supported
+func (suite *PlanTestSuite) TestDuplicatedEndpointsForSameResourceRetain() {
+	current := []*endpoint.Endpoint{suite.fooV1Cname, suite.bar192A}
+	desired := []*endpoint.Endpoint{suite.fooV1Cname, suite.fooV3CnameSameResource}
+	expectedCreate := []*endpoint.Endpoint{}
+	expectedUpdateOld := []*endpoint.Endpoint{}
+	expectedUpdateNew := []*endpoint.Endpoint{}
+	expectedDelete := []*endpoint.Endpoint{suite.bar192A}
+
+	p := &Plan{
+		Policies: []Policy{&SyncPolicy{}},
+		Current:  current,
+		Desired:  desired,
+	}
+
+	changes := p.Calculate().Changes
+	validateEntries(suite.T(), changes.Create, expectedCreate)
+	validateEntries(suite.T(), changes.UpdateNew, expectedUpdateNew)
+	validateEntries(suite.T(), changes.UpdateOld, expectedUpdateOld)
+	validateEntries(suite.T(), changes.Delete, expectedDelete)
+}
+
+func TestPlan(t *testing.T) {
+	suite.Run(t, new(PlanTestSuite))
 }
 
 // validateEntries validates that the list of entries matches expected.
 func validateEntries(t *testing.T, entries, expected []*endpoint.Endpoint) {
-	if len(entries) != len(expected) {
+	if !testutils.SameEndpoints(entries, expected) {
 		t.Fatalf("expected %q to match %q", entries, expected)
 	}
-
-	for i := range entries {
-		if !testutils.SameEndpoint(entries[i], expected[i]) {
-			t.Fatalf("expected %q to match %q", entries, expected)
-		}
-	}
-}
-
-func newEndpointWithOwner(dnsName, target, ownerID string) *endpoint.Endpoint {
-	e := endpoint.NewEndpoint(dnsName, target, endpoint.RecordTypeCNAME)
-	e.Labels[endpoint.OwnerLabelKey] = ownerID
-	return e
 }

registry/registry.go

@@ -23,7 +23,7 @@ import (
 )
 
 // Registry is an interface which should enables ownership concept in external-dns
-// Records() returns ALL records registered with DNS provider (TODO: for multi-zone support return all records)
+// Records() returns ALL records registered with DNS provider
 // each entry includes owner information
 // ApplyChanges(changes *plan.Changes) propagates the changes to the DNS Provider API and correspondingly updates ownership depending on type of registry being used
 type Registry interface {

registry/txt.go

@@ -19,8 +19,6 @@ package registry
 import (
 	"errors"
 
-	"fmt"
-	"regexp"
 	"strings"
 
 	"github.com/kubernetes-incubator/external-dns/endpoint"
@@ -28,14 +26,6 @@ import (
 	"github.com/kubernetes-incubator/external-dns/provider"
 )
 
-const (
-	txtLabelFormat = "\"heritage=external-dns,external-dns/owner=%s\""
-)
-
-var (
-	txtLabelRegex = regexp.MustCompile("^\"heritage=external-dns,external-dns/owner=(.+)\"")
-)
-
 // TXTRegistry implements registry interface with ownership implemented via associated TXT records
 type TXTRegistry struct {
 	provider provider.Provider
@@ -67,31 +57,40 @@ func (im *TXTRegistry) Records() ([]*endpoint.Endpoint, error) {
 		return nil, err
 	}
 
-	endpoints := make([]*endpoint.Endpoint, 0)
+	endpoints := []*endpoint.Endpoint{}
 
-	ownerMap := map[string]string{}
+	labelMap := map[string]endpoint.Labels{}
 
 	for _, record := range records {
 		if record.RecordType != endpoint.RecordTypeTXT {
 			endpoints = append(endpoints, record)
 			continue
 		}
-		ownerID := im.extractOwnerID(record.Target)
-		if ownerID == "" {
+		labels, err := endpoint.NewLabelsFromString(record.Target)
+		if err == endpoint.ErrInvalidHeritage {
+			//if no heritage is found or it is invalid
 			//case when value of txt record cannot be identified
 			//record will not be removed as it will have empty owner
 			endpoints = append(endpoints, record)
 			continue
 		}
+		if err != nil {
+			return nil, err
+		}
 		endpointDNSName := im.mapper.toEndpointName(record.DNSName)
-		ownerMap[endpointDNSName] = ownerID
+		labelMap[endpointDNSName] = labels
 	}
 
 	for _, ep := range endpoints {
-		ep.Labels[endpoint.OwnerLabelKey] = ownerMap[ep.DNSName]
+		if labels, ok := labelMap[ep.DNSName]; ok {
+			ep.Labels = labels
+		} else {
+			//this indicates that owner could not be identified, as there is no corresponding TXT record
+			ep.Labels = endpoint.NewLabels()
+		}
 	}
 
-	return endpoints, err
+	return endpoints, nil
 }
 
 // ApplyChanges updates dns provider with the changes
@@ -103,18 +102,33 @@ func (im *TXTRegistry) ApplyChanges(changes *plan.Changes) error {
 		UpdateOld: filterOwnedRecords(im.ownerID, changes.UpdateOld),
 		Delete:    filterOwnedRecords(im.ownerID, changes.Delete),
 	}
-
 	for _, r := range filteredChanges.Create {
-		txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), im.getTXTLabel(), endpoint.RecordTypeTXT)
+		r.Labels[endpoint.OwnerLabelKey] = im.ownerID
+		txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), r.Labels.Serialize(true), endpoint.RecordTypeTXT)
 		filteredChanges.Create = append(filteredChanges.Create, txt)
 	}
 
 	for _, r := range filteredChanges.Delete {
-		txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), im.getTXTLabel(), endpoint.RecordTypeTXT)
+		txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), r.Labels.Serialize(true), endpoint.RecordTypeTXT)
 
+		// when we delete TXT records for which value has changed (due to new label) this would still work because
+		// !!! TXT record value is uniquely generated from the Labels of the endpoint. Hence old TXT record can be uniquely reconstructed
 		filteredChanges.Delete = append(filteredChanges.Delete, txt)
 	}
 
+	// make sure TXT records are consistently updated as well
+	for _, r := range filteredChanges.UpdateNew {
+		txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), r.Labels.Serialize(true), endpoint.RecordTypeTXT)
+		filteredChanges.UpdateNew = append(filteredChanges.UpdateNew, txt)
+	}
+	// make sure TXT records are consistently updated as well
+	for _, r := range filteredChanges.UpdateOld {
+		txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), r.Labels.Serialize(true), endpoint.RecordTypeTXT)
+		// when we updateOld TXT records for which value has changed (due to new label) this would still work because
+		// !!! TXT record value is uniquely generated from the Labels of the endpoint. Hence old TXT record can be uniquely reconstructed
+		filteredChanges.UpdateOld = append(filteredChanges.UpdateOld, txt)
+	}
+
 	return im.provider.ApplyChanges(filteredChanges)
 }
 
@@ -122,17 +136,6 @@ func (im *TXTRegistry) ApplyChanges(changes *plan.Changes) error {
   TXT registry specific private methods
 */
 
-func (im *TXTRegistry) getTXTLabel() string {
-	return fmt.Sprintf(txtLabelFormat, im.ownerID)
-}
-
-func (im *TXTRegistry) extractOwnerID(txtLabel string) string {
-	if matches := txtLabelRegex.FindStringSubmatch(txtLabel); len(matches) == 2 {
-		return matches[1]
-	}
-	return ""
-}
-
 /**
   nameMapper defines interface which maps the dns name defined for the source
   to the dns name which TXT record will be created with

registry/txt_test.go

@@ -132,6 +132,7 @@ func testTXTRegistryRecordsPrefixed(t *testing.T) {
 
 	r, _ := NewTXTRegistry(p, "txt.", "owner")
 	records, _ := r.Records()
+
 	assert.True(t, testutils.SameEndpoints(records, expectedRecords))
 }
 
@@ -142,7 +143,7 @@ func testTXTRegistryRecordsNoPrefix(t *testing.T) {
 		Create: []*endpoint.Endpoint{
 			newEndpointWithOwner("foo.test-zone.example.org", "foo.loadbalancer.com", endpoint.RecordTypeCNAME, ""),
 			newEndpointWithOwner("bar.test-zone.example.org", "my-domain.com", endpoint.RecordTypeCNAME, ""),
-			newEndpointWithOwner("txt.bar.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner\"", endpoint.RecordTypeTXT, ""),
+			newEndpointWithOwner("txt.bar.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner,external-dns/resource=ingress/default/my-ingress\"", endpoint.RecordTypeTXT, ""),
 			newEndpointWithOwner("txt.bar.test-zone.example.org", "baz.test-zone.example.org", endpoint.RecordTypeCNAME, ""),
 			newEndpointWithOwner("qux.test-zone.example.org", "random", endpoint.RecordTypeTXT, ""),
 			newEndpointWithOwner("tar.test-zone.example.org", "tar.loadbalancer.com", endpoint.RecordTypeCNAME, ""),
@@ -173,7 +174,8 @@ func testTXTRegistryRecordsNoPrefix(t *testing.T) {
 			Target:     "baz.test-zone.example.org",
 			RecordType: endpoint.RecordTypeCNAME,
 			Labels: map[string]string{
-				endpoint.OwnerLabelKey: "owner",
+				endpoint.OwnerLabelKey:    "owner",
+				endpoint.ResourceLabelKey: "ingress/default/my-ingress",
 			},
 		},
 		{
@@ -233,13 +235,13 @@ func testTXTRegistryApplyChangesWithPrefix(t *testing.T) {
 
 	changes := &plan.Changes{
 		Create: []*endpoint.Endpoint{
-			newEndpointWithOwner("new-record-1.test-zone.example.org", "new-loadbalancer-1.lb.com", "", ""),
+			newEndpointWithOwnerResource("new-record-1.test-zone.example.org", "new-loadbalancer-1.lb.com", "", "", "ingress/default/my-ingress"),
 		},
 		Delete: []*endpoint.Endpoint{
 			newEndpointWithOwner("foobar.test-zone.example.org", "foobar.loadbalancer.com", endpoint.RecordTypeCNAME, "owner"),
 		},
 		UpdateNew: []*endpoint.Endpoint{
-			newEndpointWithOwner("tar.test-zone.example.org", "new-tar.loadbalancer.com", endpoint.RecordTypeCNAME, "owner"),
+			newEndpointWithOwnerResource("tar.test-zone.example.org", "new-tar.loadbalancer.com", endpoint.RecordTypeCNAME, "owner", "ingress/default/my-ingress-2"),
 		},
 		UpdateOld: []*endpoint.Endpoint{
 			newEndpointWithOwner("tar.test-zone.example.org", "tar.loadbalancer.com", endpoint.RecordTypeCNAME, "owner"),
@@ -247,18 +249,20 @@ func testTXTRegistryApplyChangesWithPrefix(t *testing.T) {
 	}
 	expected := &plan.Changes{
 		Create: []*endpoint.Endpoint{
-			newEndpointWithOwner("new-record-1.test-zone.example.org", "new-loadbalancer-1.lb.com", "", ""),
-			newEndpointWithOwner("txt.new-record-1.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner\"", endpoint.RecordTypeTXT, ""),
+			newEndpointWithOwnerResource("new-record-1.test-zone.example.org", "new-loadbalancer-1.lb.com", "", "owner", "ingress/default/my-ingress"),
+			newEndpointWithOwner("txt.new-record-1.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner,external-dns/resource=ingress/default/my-ingress\"", endpoint.RecordTypeTXT, ""),
 		},
 		Delete: []*endpoint.Endpoint{
 			newEndpointWithOwner("foobar.test-zone.example.org", "foobar.loadbalancer.com", endpoint.RecordTypeCNAME, "owner"),
 			newEndpointWithOwner("txt.foobar.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner\"", endpoint.RecordTypeTXT, ""),
 		},
 		UpdateNew: []*endpoint.Endpoint{
-			newEndpointWithOwner("tar.test-zone.example.org", "new-tar.loadbalancer.com", endpoint.RecordTypeCNAME, "owner"),
+			newEndpointWithOwnerResource("tar.test-zone.example.org", "new-tar.loadbalancer.com", endpoint.RecordTypeCNAME, "owner", "ingress/default/my-ingress-2"),
+			newEndpointWithOwner("txt.tar.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner,external-dns/resource=ingress/default/my-ingress-2\"", endpoint.RecordTypeTXT, ""),
 		},
 		UpdateOld: []*endpoint.Endpoint{
 			newEndpointWithOwner("tar.test-zone.example.org", "tar.loadbalancer.com", endpoint.RecordTypeCNAME, "owner"),
+			newEndpointWithOwner("txt.tar.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner\"", endpoint.RecordTypeTXT, ""),
 		},
 	}
 	p.OnApplyChanges = func(got *plan.Changes) {
@@ -300,7 +304,7 @@ func testTXTRegistryApplyChangesNoPrefix(t *testing.T) {
 
 	changes := &plan.Changes{
 		Create: []*endpoint.Endpoint{
-			newEndpointWithOwner("new-record-1.test-zone.example.org", "new-loadbalancer-1.lb.com", "", ""),
+			newEndpointWithOwner("new-record-1.test-zone.example.org", "new-loadbalancer-1.lb.com", endpoint.RecordTypeCNAME, ""),
 		},
 		Delete: []*endpoint.Endpoint{
 			newEndpointWithOwner("foobar.test-zone.example.org", "foobar.loadbalancer.com", endpoint.RecordTypeCNAME, "owner"),
@@ -314,7 +318,7 @@ func testTXTRegistryApplyChangesNoPrefix(t *testing.T) {
 	}
 	expected := &plan.Changes{
 		Create: []*endpoint.Endpoint{
-			newEndpointWithOwner("new-record-1.test-zone.example.org", "new-loadbalancer-1.lb.com", "", ""),
+			newEndpointWithOwner("new-record-1.test-zone.example.org", "new-loadbalancer-1.lb.com", endpoint.RecordTypeCNAME, "owner"),
 			newEndpointWithOwner("new-record-1.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner\"", endpoint.RecordTypeTXT, ""),
 		},
 		Delete: []*endpoint.Endpoint{
@@ -354,3 +358,10 @@ func newEndpointWithOwner(dnsName, target, recordType, ownerID string) *endpoint
 	e.Labels[endpoint.OwnerLabelKey] = ownerID
 	return e
 }
+
+func newEndpointWithOwnerResource(dnsName, target, recordType, ownerID, resource string) *endpoint.Endpoint {
+	e := endpoint.NewEndpoint(dnsName, target, recordType)
+	e.Labels[endpoint.OwnerLabelKey] = ownerID
+	e.Labels[endpoint.ResourceLabelKey] = resource
+	return e
+}

source/ingress.go

@@ -105,6 +105,7 @@ func (sc *ingressSource) Endpoints() ([]*endpoint.Endpoint, error) {
 		}
 
 		log.Debugf("Endpoints generated from ingress: %s/%s: %v", ing.Namespace, ing.Name, ingEndpoints)
+		sc.setResourceLabel(ing, ingEndpoints)
 		endpoints = append(endpoints, ingEndpoints...)
 	}
 
@@ -197,6 +198,12 @@ func (sc *ingressSource) filterByAnnotations(ingresses []v1beta1.Ingress) ([]v1b
 	return filteredList, nil
 }
 
+func (sc *ingressSource) setResourceLabel(ingress v1beta1.Ingress, endpoints []*endpoint.Endpoint) {
+	for _, ep := range endpoints {
+		ep.Labels[endpoint.ResourceLabelKey] = fmt.Sprintf("ingress/%s/%s", ingress.Namespace, ingress.Name)
+	}
+}
+
 // endpointsFromIngress extracts the endpoints from ingress object
 func endpointsFromIngress(ing *v1beta1.Ingress) []*endpoint.Endpoint {
 	var endpoints []*endpoint.Endpoint

source/ingress_test.go

@@ -28,12 +28,50 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
 )
 
 // Validates that ingressSource is a Source
 var _ Source = &ingressSource{}
 
+type IngressSuite struct {
+	suite.Suite
+	sc             Source
+	fooWithTargets *v1beta1.Ingress
+}
+
+func (suite *IngressSuite) SetupTest() {
+	fakeClient := fake.NewSimpleClientset()
+	var err error
+
+	suite.sc, err = NewIngressSource(
+		fakeClient,
+		"",
+		"",
+		"{{.Name}}",
+	)
+	suite.NoError(err, "should initialize ingress source")
+
+	suite.fooWithTargets = (fakeIngress{
+		name:      "foo-with-targets",
+		namespace: "default",
+		dnsnames:  []string{"foo"},
+		ips:       []string{"8.8.8.8"},
+		hostnames: []string{"v1"},
+	}).Ingress()
+	_, err = fakeClient.Extensions().Ingresses(suite.fooWithTargets.Namespace).Create(suite.fooWithTargets)
+	suite.NoError(err, "should succeed")
+}
+
+func (suite *IngressSuite) TestResourceLabelIsSet() {
+	endpoints, _ := suite.sc.Endpoints()
+	for _, ep := range endpoints {
+		suite.Equal("ingress/default/foo-with-targets", ep.Labels[endpoint.ResourceLabelKey], "should set correct resource label")
+	}
+}
+
 func TestIngress(t *testing.T) {
+	suite.Run(t, new(IngressSuite))
 	t.Run("endpointsFromIngress", testEndpointsFromIngress)
 	t.Run("Endpoints", testIngressEndpoints)
 }

source/service.go

@@ -115,6 +115,7 @@ func (sc *serviceSource) Endpoints() ([]*endpoint.Endpoint, error) {
 		}
 
 		log.Debugf("Endpoints generated from service: %s/%s: %v", svc.Namespace, svc.Name, svcEndpoints)
+		sc.setResourceLabel(svc, svcEndpoints)
 		endpoints = append(endpoints, svcEndpoints...)
 	}
 
@@ -210,6 +211,12 @@ func (sc *serviceSource) filterByAnnotations(services []v1.Service) ([]v1.Servic
 	return filteredList, nil
 }
 
+func (sc *serviceSource) setResourceLabel(service v1.Service, endpoints []*endpoint.Endpoint) {
+	for _, ep := range endpoints {
+		ep.Labels[endpoint.ResourceLabelKey] = fmt.Sprintf("service/%s/%s", service.Namespace, service.Name)
+	}
+}
+
 func (sc *serviceSource) generateEndpoints(svc *v1.Service, hostname string) []*endpoint.Endpoint {
 	var endpoints []*endpoint.Endpoint
 

source/service_test.go

@@ -28,9 +28,62 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
 )
 
+type ServiceSuite struct {
+	suite.Suite
+	sc             Source
+	fooWithTargets *v1.Service
+}
+
+func (suite *ServiceSuite) SetupTest() {
+	fakeClient := fake.NewSimpleClientset()
+	var err error
+
+	suite.sc, err = NewServiceSource(
+		fakeClient,
+		"",
+		"",
+		"{{.Name}}",
+		"",
+		false,
+	)
+	suite.fooWithTargets = &v1.Service{
+		Spec: v1.ServiceSpec{
+			Type: v1.ServiceTypeLoadBalancer,
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace:   "default",
+			Name:        "foo-with-targets",
+			Annotations: map[string]string{},
+		},
+		Status: v1.ServiceStatus{
+			LoadBalancer: v1.LoadBalancerStatus{
+				Ingress: []v1.LoadBalancerIngress{
+					{IP: "8.8.8.8"},
+					{Hostname: "foo"},
+				},
+			},
+		},
+	}
+
+	suite.NoError(err, "should initialize service source")
+
+	_, err = fakeClient.CoreV1().Services(suite.fooWithTargets.Namespace).Create(suite.fooWithTargets)
+	suite.NoError(err, "should successfully create service")
+
+}
+
+func (suite *ServiceSuite) TestResourceLabelIsSet() {
+	endpoints, _ := suite.sc.Endpoints()
+	for _, ep := range endpoints {
+		suite.Equal("service/default/foo-with-targets", ep.Labels[endpoint.ResourceLabelKey], "should set correct resource label")
+	}
+}
+
 func TestServiceSource(t *testing.T) {
+	suite.Run(t, new(ServiceSuite))
 	t.Run("Interface", testServiceSourceImplementsSource)
 	t.Run("NewServiceSource", testServiceSourceNewServiceSource)
 	t.Run("Endpoints", testServiceSourceEndpoints)