Merge pull request #1008 from devkid/feature/aws-routing-policies

[RFC] Add support for all AWS Route53 routing policies; add additional Setldentifier abstraction layer

docs/tutorials/aws.md

@@ -327,6 +327,23 @@ spec:
 
 This will set the DNS record's TTL to 60 seconds.
 
+## Routing policies
+
+Route53 offers [different routing policies](https://docs.aws.amazon.com/Route53/latest/DeveloperGuide/routing-policy.html). The routing policy for a record can be controlled with the following annotations:
+
+* `external-dns.alpha.kubernetes.io/set-identifier`: this **needs** to be set to use any of the following routing policies
+
+For any given DNS name, only **one** of the following routing policies can be used:
+
+* Weighted records: `external-dns.alpha.kubernetes.io/aws-weight`
+* Latency-based routing: `external-dns.alpha.kubernetes.io/aws-region`
+* Failover:`external-dns.alpha.kubernetes.io/aws-failover`
+* Geolocation-based routing:
+  * `external-dns.alpha.kubernetes.io/aws-geolocation-continent-code`
+  * `external-dns.alpha.kubernetes.io/aws-geolocation-country-code`
+  * `external-dns.alpha.kubernetes.io/aws-geolocation-subdivision-code`
+* Multi-value answer:`external-dns.alpha.kubernetes.io/aws-multi-value-answer`
+
 ## Clean up
 
 Make sure to delete all Service objects before terminating the cluster so all load balancers get cleaned up correctly.

endpoint/endpoint.go

@@ -126,6 +126,8 @@ type Endpoint struct {
 	Targets Targets `json:"targets,omitempty"`
 	// RecordType type of record, e.g. CNAME, A, SRV, TXT etc
 	RecordType string `json:"recordType,omitempty"`
+	// Identifier to distinguish multiple records with the same name and type (e.g. Route53 records with routing policies other than 'simple')
+	SetIdentifier string `json:"setIdentifier,omitempty"`
 	// TTL for the record
 	RecordTTL TTL `json:"recordTTL,omitempty"`
 	// Labels stores labels defined for the Endpoint
@@ -157,6 +159,11 @@ func NewEndpointWithTTL(dnsName, recordType string, ttl TTL, targets ...string)
 	}
 }
 
+func (e *Endpoint) WithSetIdentifier(setIdentifier string) *Endpoint {
+	e.SetIdentifier = setIdentifier
+	return e
+}
+
 // WithProviderSpecific attaches a key/value pair to the Endpoint and returns the Endpoint.
 // This can be used to pass additional data through the stages of ExternalDNS's Endpoint processing.
 // The assumption is that most of the time this will be provider specific metadata that doesn't
@@ -182,7 +189,7 @@ func (e *Endpoint) GetProviderSpecificProperty(key string) (ProviderSpecificProp
 }
 
 func (e *Endpoint) String() string {
-	return fmt.Sprintf("%s %d IN %s %s %s", e.DNSName, e.RecordTTL, e.RecordType, e.Targets, e.ProviderSpecific)
+	return fmt.Sprintf("%s %d IN %s %s %s %s", e.DNSName, e.RecordTTL, e.RecordType, e.SetIdentifier, e.Targets, e.ProviderSpecific)
 }
 
 // DNSEndpointSpec defines the desired state of DNSEndpoint

internal/testutils/endpoint.go

@@ -47,10 +47,10 @@ func (b byAllFields) Less(i, j int) bool {
 // SameEndpoint returns true if two endpoints are same
 // considers example.org. and example.org DNSName/Target as different endpoints
 func SameEndpoint(a, b *endpoint.Endpoint) bool {
-	return a.DNSName == b.DNSName && a.Targets.Same(b.Targets) && a.RecordType == b.RecordType &&
+	return a.DNSName == b.DNSName && a.Targets.Same(b.Targets) && a.RecordType == b.RecordType && a.SetIdentifier == b.SetIdentifier &&
 		a.Labels[endpoint.OwnerLabelKey] == b.Labels[endpoint.OwnerLabelKey] && a.RecordTTL == b.RecordTTL &&
 		a.Labels[endpoint.ResourceLabelKey] == b.Labels[endpoint.ResourceLabelKey] &&
-		SameProverSpecific(a.ProviderSpecific, b.ProviderSpecific)
+		SameProviderSpecific(a.ProviderSpecific, b.ProviderSpecific)
 }
 
 // SameEndpoints compares two slices of endpoints regardless of order
@@ -101,7 +101,7 @@ func SamePlanChanges(a, b map[string][]*endpoint.Endpoint) bool {
 		SameEndpoints(a["UpdateOld"], b["UpdateOld"]) && SameEndpoints(a["UpdateNew"], b["UpdateNew"])
 }
 
-// SameProverSpecific verifies that two maps contain the same string/string key/value pairs
+// SameProviderSpecific verifies that two maps contain the same string/string key/value pairs
-func SameProverSpecific(a, b endpoint.ProviderSpecific) bool {
+func SameProviderSpecific(a, b endpoint.ProviderSpecific) bool {
 	return reflect.DeepEqual(a, b)
 }

internal/testutils/endpoint_test.go

@@ -40,9 +40,10 @@ func ExampleSameEndpoints() {
 			RecordType: endpoint.RecordTypeTXT,
 		},
 		{
-			DNSName:    "abc.com",
+			DNSName:       "abc.com",
-			Targets:    endpoint.Targets{"1.2.3.4"},
+			Targets:       endpoint.Targets{"1.2.3.4"},
-			RecordType: endpoint.RecordTypeA,
+			RecordType:    endpoint.RecordTypeA,
+			SetIdentifier: "test-set-1",
 		},
 		{
 			DNSName:    "bbc.com",
@@ -68,11 +69,11 @@ func ExampleSameEndpoints() {
 		fmt.Println(ep)
 	}
 	// Output:
-	// abc.com 0 IN A 1.2.3.4 []
+	// abc.com 0 IN A test-set-1 1.2.3.4 []
-	// abc.com 0 IN TXT something []
+	// abc.com 0 IN TXT  something []
-	// bbc.com 0 IN CNAME foo.com []
+	// bbc.com 0 IN CNAME  foo.com []
-	// cbc.com 60 IN CNAME foo.com []
+	// cbc.com 60 IN CNAME  foo.com []
-	// example.org 0 IN  load-balancer.org []
+	// example.org 0 IN   load-balancer.org []
-	// example.org 0 IN  load-balancer.org [{foo bar}]
+	// example.org 0 IN   load-balancer.org [{foo bar}]
-	// example.org 0 IN TXT load-balancer.org []
+	// example.org 0 IN TXT  load-balancer.org []
 }

plan/plan.go

@@ -63,12 +63,12 @@ bar.com |                | [->191.1.1.1, ->190.1.1.1]  |  = create (bar.com -> 1
 "=", i.e. result of calculation relies on supplied ConflictResolver
 */
 type planTable struct {
-	rows     map[string]*planTableRow
+	rows     map[string]map[string]*planTableRow
 	resolver ConflictResolver
 }
 
 func newPlanTable() planTable { //TODO: make resolver configurable
-	return planTable{map[string]*planTableRow{}, PerResource{}}
+	return planTable{map[string]map[string]*planTableRow{}, PerResource{}}
 }
 
 // planTableRow
@@ -86,52 +86,23 @@ func (t planTableRow) String() string {
 func (t planTable) addCurrent(e *endpoint.Endpoint) {
 	dnsName := normalizeDNSName(e.DNSName)
 	if _, ok := t.rows[dnsName]; !ok {
-		t.rows[dnsName] = &planTableRow{}
+		t.rows[dnsName] = make(map[string]*planTableRow)
 	}
-	t.rows[dnsName].current = e
+	if _, ok := t.rows[dnsName][e.SetIdentifier]; !ok {
+		t.rows[dnsName][e.SetIdentifier] = &planTableRow{}
+	}
+	t.rows[dnsName][e.SetIdentifier].current = e
 }
 
 func (t planTable) addCandidate(e *endpoint.Endpoint) {
 	dnsName := normalizeDNSName(e.DNSName)
 	if _, ok := t.rows[dnsName]; !ok {
-		t.rows[dnsName] = &planTableRow{}
+		t.rows[dnsName] = make(map[string]*planTableRow)
 	}
-	t.rows[dnsName].candidates = append(t.rows[dnsName].candidates, e)
+	if _, ok := t.rows[dnsName][e.SetIdentifier]; !ok {
-}
+		t.rows[dnsName][e.SetIdentifier] = &planTableRow{}
-
-// 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) || shouldUpdateProviderSpecific(update, row.current) {
-				inheritOwner(row.current, update)
-				updateNew = append(updateNew, update)
-				updateOld = append(updateOld, row.current)
-			}
-			continue
-		}
 	}
-	return
+	t.rows[dnsName][e.SetIdentifier].candidates = append(t.rows[dnsName][e.SetIdentifier].candidates, e)
-}
-
-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
@@ -148,9 +119,29 @@ func (p *Plan) Calculate() *Plan {
 	}
 
 	changes := &Changes{}
-	changes.Create = t.getCreates()
+
-	changes.Delete = t.getDeletes()
+	for _, topRow := range t.rows {
-	changes.UpdateNew, changes.UpdateOld = t.getUpdates()
+		for _, row := range topRow {
+			if row.current == nil { //dns name not taken
+				changes.Create = append(changes.Create, t.resolver.ResolveCreate(row.candidates))
+			}
+			if row.current != nil && len(row.candidates) == 0 {
+				changes.Delete = append(changes.Delete, row.current)
+			}
+
+			// TODO: allows record type change, which might not be supported by all dns providers
+			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) || shouldUpdateProviderSpecific(update, row.current) {
+					inheritOwner(row.current, update)
+					changes.UpdateNew = append(changes.UpdateNew, update)
+					changes.UpdateOld = append(changes.UpdateOld, row.current)
+				}
+				continue
+			}
+		}
+	}
 	for _, pol := range p.Policies {
 		changes = pol.Apply(changes)
 	}
@@ -196,11 +187,34 @@ func shouldUpdateProviderSpecific(desired, current *endpoint.Endpoint) bool {
 			continue
 		}
 
+		found := false
 		for _, d := range desired.ProviderSpecific {
-			if d.Name == c.Name && d.Value != c.Value {
+			if d.Name == c.Name {
-				return true
+				if d.Value != c.Value {
+					// provider-specific attribute updated
+					return true
+				}
+				found = true
+				break
 			}
 		}
+		if !found {
+			// provider-specific attribute deleted
+			return true
+		}
+	}
+	for _, d := range desired.ProviderSpecific {
+		found := false
+		for _, c := range current.ProviderSpecific {
+			if d.Name == c.Name {
+				found = true
+				break
+			}
+		}
+		if !found {
+			// provider-specific attribute added
+			return true
+		}
 	}
 
 	return false

plan/plan_test.go

@@ -38,6 +38,9 @@ type PlanTestSuite struct {
 	bar127AWithProviderSpecificTrue  *endpoint.Endpoint
 	bar127AWithProviderSpecificFalse *endpoint.Endpoint
 	bar192A                          *endpoint.Endpoint
+	multiple1                        *endpoint.Endpoint
+	multiple2                        *endpoint.Endpoint
+	multiple3                        *endpoint.Endpoint
 }
 
 func (suite *PlanTestSuite) SetupTest() {
@@ -138,7 +141,24 @@ func (suite *PlanTestSuite) SetupTest() {
 			endpoint.ResourceLabelKey: "ingress/default/bar-192",
 		},
 	}
-
+	suite.multiple1 = &endpoint.Endpoint{
+		DNSName:       "multiple",
+		Targets:       endpoint.Targets{"192.168.0.1"},
+		RecordType:    "A",
+		SetIdentifier: "test-set-1",
+	}
+	suite.multiple2 = &endpoint.Endpoint{
+		DNSName:       "multiple",
+		Targets:       endpoint.Targets{"192.168.0.2"},
+		RecordType:    "A",
+		SetIdentifier: "test-set-1",
+	}
+	suite.multiple3 = &endpoint.Endpoint{
+		DNSName:       "multiple",
+		Targets:       endpoint.Targets{"192.168.0.2"},
+		RecordType:    "A",
+		SetIdentifier: "test-set-2",
+	}
 }
 
 func (suite *PlanTestSuite) TestSyncFirstRound() {
@@ -427,6 +447,50 @@ func (suite *PlanTestSuite) TestDuplicatedEndpointsForSameResourceRetain() {
 	validateEntries(suite.T(), changes.Delete, expectedDelete)
 }
 
+func (suite *PlanTestSuite) TestMultipleRecordsSameNameDifferentSetIdentifier() {
+
+	current := []*endpoint.Endpoint{suite.multiple1}
+	desired := []*endpoint.Endpoint{suite.multiple2, suite.multiple3}
+	expectedCreate := []*endpoint.Endpoint{suite.multiple3}
+	expectedUpdateOld := []*endpoint.Endpoint{suite.multiple1}
+	expectedUpdateNew := []*endpoint.Endpoint{suite.multiple2}
+	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) TestSetIdentifierUpdateCreatesAndDeletes() {
+
+	current := []*endpoint.Endpoint{suite.multiple2}
+	desired := []*endpoint.Endpoint{suite.multiple3}
+	expectedCreate := []*endpoint.Endpoint{suite.multiple3}
+	expectedUpdateOld := []*endpoint.Endpoint{}
+	expectedUpdateNew := []*endpoint.Endpoint{}
+	expectedDelete := []*endpoint.Endpoint{suite.multiple2}
+
+	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))
 }

provider/aws.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"sort"
+	"strconv"
 	"strings"
 	"time"
 
@@ -37,7 +38,14 @@ const (
 	recordTTL = 300
 	// provider specific key that designates whether an AWS ALIAS record has the EvaluateTargetHealth
 	// field set to true.
-	providerSpecificEvaluateTargetHealth = "aws/evaluate-target-health"
+	providerSpecificEvaluateTargetHealth       = "aws/evaluate-target-health"
+	providerSpecificWeight                     = "aws/weight"
+	providerSpecificRegion                     = "aws/region"
+	providerSpecificFailover                   = "aws/failover"
+	providerSpecificGeolocationContinentCode   = "aws/geolocation-continent-code"
+	providerSpecificGeolocationCountryCode     = "aws/geolocation-country-code"
+	providerSpecificGeolocationSubdivisionCode = "aws/geolocation-subdivision-code"
+	providerSpecificMultiValueAnswer           = "aws/multi-value-answer"
 )
 
 var (
@@ -249,6 +257,8 @@ func (p *AWSProvider) records(zones map[string]*route53.HostedZone) ([]*endpoint
 	endpoints := make([]*endpoint.Endpoint, 0)
 	f := func(resp *route53.ListResourceRecordSetsOutput, lastPage bool) (shouldContinue bool) {
 		for _, r := range resp.ResourceRecordSets {
+			newEndpoints := make([]*endpoint.Endpoint, 0)
+
 			// TODO(linki, ownership): Remove once ownership system is in place.
 			// See: https://github.com/kubernetes-sigs/external-dns/pull/122/files/74e2c3d3e237411e619aefc5aab694742001cdec#r109863370
 
@@ -267,7 +277,7 @@ func (p *AWSProvider) records(zones map[string]*route53.HostedZone) ([]*endpoint
 					targets[idx] = aws.StringValue(rr.Value)
 				}
 
-				endpoints = append(endpoints, endpoint.NewEndpointWithTTL(wildcardUnescape(aws.StringValue(r.Name)), aws.StringValue(r.Type), ttl, targets...))
+				newEndpoints = append(newEndpoints, endpoint.NewEndpointWithTTL(wildcardUnescape(aws.StringValue(r.Name)), aws.StringValue(r.Type), ttl, targets...))
 			}
 
 			if r.AliasTarget != nil {
@@ -278,6 +288,36 @@ func (p *AWSProvider) records(zones map[string]*route53.HostedZone) ([]*endpoint
 				ep := endpoint.
 					NewEndpointWithTTL(wildcardUnescape(aws.StringValue(r.Name)), endpoint.RecordTypeCNAME, ttl, aws.StringValue(r.AliasTarget.DNSName)).
 					WithProviderSpecific(providerSpecificEvaluateTargetHealth, fmt.Sprintf("%t", aws.BoolValue(r.AliasTarget.EvaluateTargetHealth)))
+				newEndpoints = append(newEndpoints, ep)
+			}
+
+			for _, ep := range newEndpoints {
+				if r.SetIdentifier != nil {
+					ep.SetIdentifier = aws.StringValue(r.SetIdentifier)
+					switch {
+					case r.Weight != nil:
+						ep.WithProviderSpecific(providerSpecificWeight, fmt.Sprintf("%d", aws.Int64Value(r.Weight)))
+					case r.Region != nil:
+						ep.WithProviderSpecific(providerSpecificRegion, aws.StringValue(r.Region))
+					case r.Failover != nil:
+						ep.WithProviderSpecific(providerSpecificFailover, aws.StringValue(r.Failover))
+					case r.MultiValueAnswer != nil && aws.BoolValue(r.MultiValueAnswer):
+						ep.WithProviderSpecific(providerSpecificMultiValueAnswer, "")
+					case r.GeoLocation != nil:
+						if r.GeoLocation.ContinentCode != nil {
+							ep.WithProviderSpecific(providerSpecificGeolocationContinentCode, aws.StringValue(r.GeoLocation.ContinentCode))
+						} else {
+							if r.GeoLocation.CountryCode != nil {
+								ep.WithProviderSpecific(providerSpecificGeolocationCountryCode, aws.StringValue(r.GeoLocation.CountryCode))
+							}
+							if r.GeoLocation.SubdivisionCode != nil {
+								ep.WithProviderSpecific(providerSpecificGeolocationSubdivisionCode, aws.StringValue(r.GeoLocation.SubdivisionCode))
+							}
+						}
+					default:
+						// one of the above needs to be set, otherwise SetIdentifier doesn't make sense
+					}
+				}
 				endpoints = append(endpoints, ep)
 			}
 		}
@@ -483,6 +523,47 @@ func (p *AWSProvider) newChange(action string, ep *endpoint.Endpoint, recordsCac
 		}
 	}
 
+	setIdentifier := ep.SetIdentifier
+	if setIdentifier != "" {
+		change.ResourceRecordSet.SetIdentifier = aws.String(setIdentifier)
+		if prop, ok := ep.GetProviderSpecificProperty(providerSpecificWeight); ok {
+			weight, err := strconv.ParseInt(prop.Value, 10, 64)
+			if err != nil {
+				log.Errorf("Failed parsing value of %s: %s: %v; using weight of 0", providerSpecificWeight, prop.Value, err)
+				weight = 0
+			}
+			change.ResourceRecordSet.Weight = aws.Int64(weight)
+		}
+		if prop, ok := ep.GetProviderSpecificProperty(providerSpecificRegion); ok {
+			change.ResourceRecordSet.Region = aws.String(prop.Value)
+		}
+		if prop, ok := ep.GetProviderSpecificProperty(providerSpecificFailover); ok {
+			change.ResourceRecordSet.Failover = aws.String(prop.Value)
+		}
+		if _, ok := ep.GetProviderSpecificProperty(providerSpecificMultiValueAnswer); ok {
+			change.ResourceRecordSet.MultiValueAnswer = aws.Bool(true)
+		}
+
+		var geolocation = &route53.GeoLocation{}
+		useGeolocation := false
+		if prop, ok := ep.GetProviderSpecificProperty(providerSpecificGeolocationContinentCode); ok {
+			geolocation.ContinentCode = aws.String(prop.Value)
+			useGeolocation = true
+		} else {
+			if prop, ok := ep.GetProviderSpecificProperty(providerSpecificGeolocationCountryCode); ok {
+				geolocation.CountryCode = aws.String(prop.Value)
+				useGeolocation = true
+			}
+			if prop, ok := ep.GetProviderSpecificProperty(providerSpecificGeolocationSubdivisionCode); ok {
+				geolocation.SubdivisionCode = aws.String(prop.Value)
+				useGeolocation = true
+			}
+		}
+		if useGeolocation {
+			change.ResourceRecordSet.GeoLocation = geolocation
+		}
+	}
+
 	return change, dualstack
 }
 

provider/aws_test.go

@@ -184,7 +184,11 @@ func (r *Route53APIStub) ChangeResourceRecordSets(input *route53.ChangeResourceR
 			change.ResourceRecordSet.AliasTarget.DNSName = aws.String(wildcardEscape(ensureTrailingDot(aws.StringValue(change.ResourceRecordSet.AliasTarget.DNSName))))
 		}
 
-		key := aws.StringValue(change.ResourceRecordSet.Name) + "::" + aws.StringValue(change.ResourceRecordSet.Type)
+		setId := ""
+		if change.ResourceRecordSet.SetIdentifier != nil {
+			setId = aws.StringValue(change.ResourceRecordSet.SetIdentifier)
+		}
+		key := aws.StringValue(change.ResourceRecordSet.Name) + "::" + aws.StringValue(change.ResourceRecordSet.Type) + "::" + setId
 		switch aws.StringValue(change.Action) {
 		case route53.ChangeActionCreate:
 			if _, found := recordSets[key]; found {
@@ -314,6 +318,13 @@ func TestAWSRecords(t *testing.T) {
 		endpoint.NewEndpoint("list-test-alias-evaluate.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeCNAME, "foo.eu-central-1.elb.amazonaws.com").WithProviderSpecific(providerSpecificEvaluateTargetHealth, "true"),
 		endpoint.NewEndpointWithTTL("list-test-multiple.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeA, endpoint.TTL(recordTTL), "8.8.8.8", "8.8.4.4"),
 		endpoint.NewEndpointWithTTL("prefix-*.wildcard.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeTXT, endpoint.TTL(recordTTL), "random"),
+		endpoint.NewEndpointWithTTL("weight-test.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeA, endpoint.TTL(recordTTL), "1.2.3.4").WithSetIdentifier("test-set-1").WithProviderSpecific(providerSpecificWeight, "10"),
+		endpoint.NewEndpointWithTTL("weight-test.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeA, endpoint.TTL(recordTTL), "4.3.2.1").WithSetIdentifier("test-set-2").WithProviderSpecific(providerSpecificWeight, "20"),
+		endpoint.NewEndpointWithTTL("latency-test.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeA, endpoint.TTL(recordTTL), "1.2.3.4").WithSetIdentifier("test-set").WithProviderSpecific(providerSpecificRegion, "us-east-1"),
+		endpoint.NewEndpointWithTTL("failover-test.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeA, endpoint.TTL(recordTTL), "1.2.3.4").WithSetIdentifier("test-set").WithProviderSpecific(providerSpecificFailover, "PRIMARY"),
+		endpoint.NewEndpointWithTTL("multi-value-answer-test.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeA, endpoint.TTL(recordTTL), "1.2.3.4").WithSetIdentifier("test-set").WithProviderSpecific(providerSpecificMultiValueAnswer, ""),
+		endpoint.NewEndpointWithTTL("geolocation-test.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeA, endpoint.TTL(recordTTL), "1.2.3.4").WithSetIdentifier("test-set-1").WithProviderSpecific(providerSpecificGeolocationContinentCode, "EU"),
+		endpoint.NewEndpointWithTTL("geolocation-test.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeA, endpoint.TTL(recordTTL), "4.3.2.1").WithSetIdentifier("test-set-2").WithProviderSpecific(providerSpecificGeolocationCountryCode, "DE"),
 	})
 
 	records, err := provider.Records()
@@ -328,6 +339,13 @@ func TestAWSRecords(t *testing.T) {
 		endpoint.NewEndpointWithTTL("list-test-alias-evaluate.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeCNAME, endpoint.TTL(recordTTL), "foo.eu-central-1.elb.amazonaws.com").WithProviderSpecific(providerSpecificEvaluateTargetHealth, "true"),
 		endpoint.NewEndpointWithTTL("list-test-multiple.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeA, endpoint.TTL(recordTTL), "8.8.8.8", "8.8.4.4"),
 		endpoint.NewEndpointWithTTL("prefix-*.wildcard.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeTXT, endpoint.TTL(recordTTL), "random"),
+		endpoint.NewEndpointWithTTL("weight-test.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeA, endpoint.TTL(recordTTL), "1.2.3.4").WithSetIdentifier("test-set-1").WithProviderSpecific(providerSpecificWeight, "10"),
+		endpoint.NewEndpointWithTTL("weight-test.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeA, endpoint.TTL(recordTTL), "4.3.2.1").WithSetIdentifier("test-set-2").WithProviderSpecific(providerSpecificWeight, "20"),
+		endpoint.NewEndpointWithTTL("latency-test.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeA, endpoint.TTL(recordTTL), "1.2.3.4").WithSetIdentifier("test-set").WithProviderSpecific(providerSpecificRegion, "us-east-1"),
+		endpoint.NewEndpointWithTTL("failover-test.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeA, endpoint.TTL(recordTTL), "1.2.3.4").WithSetIdentifier("test-set").WithProviderSpecific(providerSpecificFailover, "PRIMARY"),
+		endpoint.NewEndpointWithTTL("multi-value-answer-test.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeA, endpoint.TTL(recordTTL), "1.2.3.4").WithSetIdentifier("test-set").WithProviderSpecific(providerSpecificMultiValueAnswer, ""),
+		endpoint.NewEndpointWithTTL("geolocation-test.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeA, endpoint.TTL(recordTTL), "1.2.3.4").WithSetIdentifier("test-set-1").WithProviderSpecific(providerSpecificGeolocationContinentCode, "EU"),
+		endpoint.NewEndpointWithTTL("geolocation-test.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeA, endpoint.TTL(recordTTL), "4.3.2.1").WithSetIdentifier("test-set-2").WithProviderSpecific(providerSpecificGeolocationCountryCode, "DE"),
 	})
 }
 
@@ -797,7 +815,7 @@ func TestAWSBatchChangeSetExceedingNameChange(t *testing.T) {
 }
 
 func validateEndpoints(t *testing.T, endpoints []*endpoint.Endpoint, expected []*endpoint.Endpoint) {
-	assert.True(t, testutils.SameEndpoints(endpoints, expected), "expected and actual endpoints don't match. %s:%s", endpoints, expected)
+	assert.True(t, testutils.SameEndpoints(endpoints, expected), "actual and expected endpoints don't match. %s:%s", endpoints, expected)
 }
 
 func validateAWSZones(t *testing.T, zones map[string]*route53.HostedZone, expected map[string]*route53.HostedZone) {

provider/inmemory.go

@@ -131,7 +131,7 @@ func (im *InMemoryProvider) Records() ([]*endpoint.Endpoint, error) {
 		}
 
 		for _, record := range records {
-			ep := endpoint.NewEndpoint(record.Name, record.Type, record.Target)
+			ep := endpoint.NewEndpoint(record.Name, record.Type, record.Target).WithSetIdentifier(record.SetIdentifier)
 			ep.Labels = record.Labels
 			endpoints = append(endpoints, ep)
 		}
@@ -204,10 +204,11 @@ func convertToInMemoryRecord(endpoints []*endpoint.Endpoint) []*inMemoryRecord {
 	records := []*inMemoryRecord{}
 	for _, ep := range endpoints {
 		records = append(records, &inMemoryRecord{
-			Type:   ep.RecordType,
+			Type:          ep.RecordType,
-			Name:   ep.DNSName,
+			Name:          ep.DNSName,
-			Target: ep.Targets[0],
+			Target:        ep.Targets[0],
-			Labels: ep.Labels,
+			SetIdentifier: ep.SetIdentifier,
+			Labels:        ep.Labels,
 		})
 	}
 	return records
@@ -246,10 +247,11 @@ func (f *filter) EndpointZoneID(endpoint *endpoint.Endpoint, zones map[string]st
 // Name - DNS name assigned to the record
 // Target - target of the record
 type inMemoryRecord struct {
-	Type   string
+	Type          string
-	Name   string
+	SetIdentifier string
-	Target string
+	Name          string
-	Labels endpoint.Labels
+	Target        string
+	Labels        endpoint.Labels
 }
 
 type zone map[string][]*inMemoryRecord
@@ -328,15 +330,19 @@ func (c *inMemoryClient) ApplyChanges(ctx context.Context, zoneID string, change
 	return nil
 }
 
-func (c *inMemoryClient) updateMesh(mesh map[string]map[string]bool, record *inMemoryRecord) error {
+func (c *inMemoryClient) updateMesh(mesh map[string]map[string]map[string]bool, record *inMemoryRecord) error {
 	if _, exists := mesh[record.Name]; exists {
-		if mesh[record.Name][record.Type] {
+		if _, exists := mesh[record.Name][record.Type]; exists {
-			return ErrDuplicateRecordFound
+			if mesh[record.Name][record.Type][record.SetIdentifier] {
+				return ErrDuplicateRecordFound
+			}
+			mesh[record.Name][record.Type][record.SetIdentifier] = true
+			return nil
 		}
-		mesh[record.Name][record.Type] = true
+		mesh[record.Name][record.Type] = map[string]bool{record.SetIdentifier: true}
 		return nil
 	}
-	mesh[record.Name] = map[string]bool{record.Type: true}
+	mesh[record.Name] = map[string]map[string]bool{record.Type: {record.SetIdentifier: true}}
 	return nil
 }
 
@@ -346,9 +352,9 @@ func (c *inMemoryClient) validateChangeBatch(zone string, changes *inMemoryChang
 	if !ok {
 		return ErrZoneNotFound
 	}
-	mesh := map[string]map[string]bool{}
+	mesh := map[string]map[string]map[string]bool{}
 	for _, newEndpoint := range changes.Create {
-		if c.findByType(newEndpoint.Type, curZone[newEndpoint.Name]) != nil {
+		if c.findByTypeAndSetIdentifier(newEndpoint.Type, newEndpoint.SetIdentifier, curZone[newEndpoint.Name]) != nil {
 			return ErrRecordAlreadyExists
 		}
 		if err := c.updateMesh(mesh, newEndpoint); err != nil {
@@ -356,7 +362,7 @@ func (c *inMemoryClient) validateChangeBatch(zone string, changes *inMemoryChang
 		}
 	}
 	for _, updateEndpoint := range changes.UpdateNew {
-		if c.findByType(updateEndpoint.Type, curZone[updateEndpoint.Name]) == nil {
+		if c.findByTypeAndSetIdentifier(updateEndpoint.Type, updateEndpoint.SetIdentifier, curZone[updateEndpoint.Name]) == nil {
 			return ErrRecordNotFound
 		}
 		if err := c.updateMesh(mesh, updateEndpoint); err != nil {
@@ -364,12 +370,12 @@ func (c *inMemoryClient) validateChangeBatch(zone string, changes *inMemoryChang
 		}
 	}
 	for _, updateOldEndpoint := range changes.UpdateOld {
-		if rec := c.findByType(updateOldEndpoint.Type, curZone[updateOldEndpoint.Name]); rec == nil || rec.Target != updateOldEndpoint.Target {
+		if rec := c.findByTypeAndSetIdentifier(updateOldEndpoint.Type, updateOldEndpoint.SetIdentifier, curZone[updateOldEndpoint.Name]); rec == nil || rec.Target != updateOldEndpoint.Target {
 			return ErrRecordNotFound
 		}
 	}
 	for _, deleteEndpoint := range changes.Delete {
-		if rec := c.findByType(deleteEndpoint.Type, curZone[deleteEndpoint.Name]); rec == nil || rec.Target != deleteEndpoint.Target {
+		if rec := c.findByTypeAndSetIdentifier(deleteEndpoint.Type, deleteEndpoint.SetIdentifier, curZone[deleteEndpoint.Name]); rec == nil || rec.Target != deleteEndpoint.Target {
 			return ErrRecordNotFound
 		}
 		if err := c.updateMesh(mesh, deleteEndpoint); err != nil {
@@ -379,9 +385,9 @@ func (c *inMemoryClient) validateChangeBatch(zone string, changes *inMemoryChang
 	return nil
 }
 
-func (c *inMemoryClient) findByType(recordType string, records []*inMemoryRecord) *inMemoryRecord {
+func (c *inMemoryClient) findByTypeAndSetIdentifier(recordType, setIdentifier string, records []*inMemoryRecord) *inMemoryRecord {
 	for _, record := range records {
-		if record.Type == recordType {
+		if record.Type == recordType && record.SetIdentifier == setIdentifier {
 			return record
 		}
 	}

provider/inmemory_test.go

@@ -43,11 +43,12 @@ func TestInMemoryProvider(t *testing.T) {
 
 func testInMemoryFindByType(t *testing.T) {
 	for _, ti := range []struct {
-		title         string
+		title             string
-		findType      string
+		findType          string
-		records       []*inMemoryRecord
+		findSetIdentifier string
-		expected      *inMemoryRecord
+		records           []*inMemoryRecord
-		expectedEmpty bool
+		expected          *inMemoryRecord
+		expectedEmpty     bool
 	}{
 		{
 			title:         "no records, empty type",
@@ -112,10 +113,32 @@ func testInMemoryFindByType(t *testing.T) {
 				Type: endpoint.RecordTypeA,
 			},
 		},
+		{
+			title:             "multiple records, right type and set identifier",
+			findType:          endpoint.RecordTypeA,
+			findSetIdentifier: "test-set-1",
+			records: []*inMemoryRecord{
+				{
+					Type:          endpoint.RecordTypeA,
+					SetIdentifier: "test-set-1",
+				},
+				{
+					Type:          endpoint.RecordTypeA,
+					SetIdentifier: "test-set-2",
+				},
+				{
+					Type: endpoint.RecordTypeTXT,
+				},
+			},
+			expected: &inMemoryRecord{
+				Type:          endpoint.RecordTypeA,
+				SetIdentifier: "test-set-1",
+			},
+		},
 	} {
 		t.Run(ti.title, func(t *testing.T) {
 			c := newInMemoryClient()
-			record := c.findByType(ti.findType, ti.records)
+			record := c.findByTypeAndSetIdentifier(ti.findType, ti.findSetIdentifier, ti.records)
 			if ti.expectedEmpty {
 				assert.Nil(t, record)
 			} else {

registry/txt.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"time"
 
+	"fmt"
 	"strings"
 
 	"github.com/kubernetes-sigs/external-dns/endpoint"
@@ -94,15 +95,16 @@ func (im *TXTRegistry) Records() ([]*endpoint.Endpoint, error) {
 		if err != nil {
 			return nil, err
 		}
-		endpointDNSName := im.mapper.toEndpointName(record.DNSName)
+		key := fmt.Sprintf("%s::%s", im.mapper.toEndpointName(record.DNSName), record.SetIdentifier)
-		labelMap[endpointDNSName] = labels
+		labelMap[key] = labels
 	}
 
 	for _, ep := range endpoints {
 		if ep.Labels == nil {
 			ep.Labels = endpoint.NewLabels()
 		}
-		if labels, ok := labelMap[ep.DNSName]; ok {
+		key := fmt.Sprintf("%s::%s", ep.DNSName, ep.SetIdentifier)
+		if labels, ok := labelMap[key]; ok {
 			for k, v := range labels {
 				ep.Labels[k] = v
 			}
@@ -132,7 +134,8 @@ func (im *TXTRegistry) ApplyChanges(ctx context.Context, changes *plan.Changes)
 			r.Labels = make(map[string]string)
 		}
 		r.Labels[endpoint.OwnerLabelKey] = im.ownerID
-		txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), endpoint.RecordTypeTXT, r.Labels.Serialize(true))
+		txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), endpoint.RecordTypeTXT, r.Labels.Serialize(true)).WithSetIdentifier(r.SetIdentifier)
+		txt.ProviderSpecific = r.ProviderSpecific
 		filteredChanges.Create = append(filteredChanges.Create, txt)
 
 		if im.cacheInterval > 0 {
@@ -141,7 +144,8 @@ func (im *TXTRegistry) ApplyChanges(ctx context.Context, changes *plan.Changes)
 	}
 
 	for _, r := range filteredChanges.Delete {
-		txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), endpoint.RecordTypeTXT, r.Labels.Serialize(true))
+		txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), endpoint.RecordTypeTXT, r.Labels.Serialize(true)).WithSetIdentifier(r.SetIdentifier)
+		txt.ProviderSpecific = r.ProviderSpecific
 
 		// 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
@@ -154,7 +158,8 @@ func (im *TXTRegistry) ApplyChanges(ctx context.Context, changes *plan.Changes)
 
 	// make sure TXT records are consistently updated as well
 	for _, r := range filteredChanges.UpdateOld {
-		txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), endpoint.RecordTypeTXT, r.Labels.Serialize(true))
+		txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), endpoint.RecordTypeTXT, r.Labels.Serialize(true)).WithSetIdentifier(r.SetIdentifier)
+		txt.ProviderSpecific = r.ProviderSpecific
 		// 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)
@@ -166,7 +171,8 @@ func (im *TXTRegistry) ApplyChanges(ctx context.Context, changes *plan.Changes)
 
 	// make sure TXT records are consistently updated as well
 	for _, r := range filteredChanges.UpdateNew {
-		txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), endpoint.RecordTypeTXT, r.Labels.Serialize(true))
+		txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), endpoint.RecordTypeTXT, r.Labels.Serialize(true)).WithSetIdentifier(r.SetIdentifier)
+		txt.ProviderSpecific = r.ProviderSpecific
 		filteredChanges.UpdateNew = append(filteredChanges.UpdateNew, txt)
 		// add new version of record to cache
 		if im.cacheInterval > 0 {
@@ -230,7 +236,7 @@ func (im *TXTRegistry) removeFromCache(ep *endpoint.Endpoint) {
 	}
 
 	for i, e := range im.recordsCache {
-		if e.DNSName == ep.DNSName && e.RecordType == ep.RecordType && e.Targets.Same(ep.Targets) {
+		if e.DNSName == ep.DNSName && e.RecordType == ep.RecordType && e.SetIdentifier == ep.SetIdentifier && e.Targets.Same(ep.Targets) {
 			// We found a match delete the endpoint from the cache.
 			im.recordsCache = append(im.recordsCache[:i], im.recordsCache[i+1:]...)
 			return

registry/txt_test.go

@@ -80,6 +80,10 @@ func testTXTRegistryRecordsPrefixed(t *testing.T) {
 			newEndpointWithOwner("TxT.tar.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner-2\"", endpoint.RecordTypeTXT, ""), // case-insensitive TXT prefix
 			newEndpointWithOwner("foobar.test-zone.example.org", "foobar.loadbalancer.com", endpoint.RecordTypeCNAME, ""),
 			newEndpointWithOwner("foobar.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner\"", endpoint.RecordTypeTXT, ""),
+			newEndpointWithOwner("multiple.test-zone.example.org", "lb1.loadbalancer.com", endpoint.RecordTypeCNAME, "").WithSetIdentifier("test-set-1"),
+			newEndpointWithOwner("multiple.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner\"", endpoint.RecordTypeTXT, "").WithSetIdentifier("test-set-1"),
+			newEndpointWithOwner("multiple.test-zone.example.org", "lb2.loadbalancer.com", endpoint.RecordTypeCNAME, "").WithSetIdentifier("test-set-2"),
+			newEndpointWithOwner("multiple.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner\"", endpoint.RecordTypeTXT, "").WithSetIdentifier("test-set-2"),
 		},
 	})
 	expectedRecords := []*endpoint.Endpoint{
@@ -134,6 +138,24 @@ func testTXTRegistryRecordsPrefixed(t *testing.T) {
 				endpoint.OwnerLabelKey: "",
 			},
 		},
+		{
+			DNSName:       "multiple.test-zone.example.org",
+			Targets:       endpoint.Targets{"lb1.loadbalancer.com"},
+			SetIdentifier: "test-set-1",
+			RecordType:    endpoint.RecordTypeCNAME,
+			Labels: map[string]string{
+				endpoint.OwnerLabelKey: "",
+			},
+		},
+		{
+			DNSName:       "multiple.test-zone.example.org",
+			Targets:       endpoint.Targets{"lb2.loadbalancer.com"},
+			SetIdentifier: "test-set-2",
+			RecordType:    endpoint.RecordTypeCNAME,
+			Labels: map[string]string{
+				endpoint.OwnerLabelKey: "",
+			},
+		},
 	}
 
 	r, _ := NewTXTRegistry(p, "txt.", "owner", time.Hour)
@@ -246,6 +268,10 @@ func testTXTRegistryApplyChangesWithPrefix(t *testing.T) {
 			newEndpointWithOwner("txt.tar.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner\"", endpoint.RecordTypeTXT, ""),
 			newEndpointWithOwner("foobar.test-zone.example.org", "foobar.loadbalancer.com", endpoint.RecordTypeCNAME, ""),
 			newEndpointWithOwner("txt.foobar.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner\"", endpoint.RecordTypeTXT, ""),
+			newEndpointWithOwner("multiple.test-zone.example.org", "lb1.loadbalancer.com", endpoint.RecordTypeCNAME, "").WithSetIdentifier("test-set-1"),
+			newEndpointWithOwner("txt.multiple.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner\"", endpoint.RecordTypeTXT, "").WithSetIdentifier("test-set-1"),
+			newEndpointWithOwner("multiple.test-zone.example.org", "lb2.loadbalancer.com", endpoint.RecordTypeCNAME, "").WithSetIdentifier("test-set-2"),
+			newEndpointWithOwner("txt.multiple.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner\"", endpoint.RecordTypeTXT, "").WithSetIdentifier("test-set-2"),
 		},
 	})
 	r, _ := NewTXTRegistry(p, "txt.", "owner", time.Hour)
@@ -253,33 +279,45 @@ func testTXTRegistryApplyChangesWithPrefix(t *testing.T) {
 	changes := &plan.Changes{
 		Create: []*endpoint.Endpoint{
 			newEndpointWithOwnerResource("new-record-1.test-zone.example.org", "new-loadbalancer-1.lb.com", "", "", "ingress/default/my-ingress"),
+			newEndpointWithOwnerResource("multiple.test-zone.example.org", "lb3.loadbalancer.com", "", "", "ingress/default/my-ingress").WithSetIdentifier("test-set-3"),
 		},
 		Delete: []*endpoint.Endpoint{
 			newEndpointWithOwner("foobar.test-zone.example.org", "foobar.loadbalancer.com", endpoint.RecordTypeCNAME, "owner"),
+			newEndpointWithOwner("multiple.test-zone.example.org", "lb1.loadbalancer.com", endpoint.RecordTypeCNAME, "owner").WithSetIdentifier("test-set-1"),
 		},
 		UpdateNew: []*endpoint.Endpoint{
 			newEndpointWithOwnerResource("tar.test-zone.example.org", "new-tar.loadbalancer.com", endpoint.RecordTypeCNAME, "owner", "ingress/default/my-ingress-2"),
+			newEndpointWithOwnerResource("multiple.test-zone.example.org", "new.loadbalancer.com", endpoint.RecordTypeCNAME, "owner", "ingress/default/my-ingress-2").WithSetIdentifier("test-set-2"),
 		},
 		UpdateOld: []*endpoint.Endpoint{
 			newEndpointWithOwner("tar.test-zone.example.org", "tar.loadbalancer.com", endpoint.RecordTypeCNAME, "owner"),
+			newEndpointWithOwner("multiple.test-zone.example.org", "lb2.loadbalancer.com", endpoint.RecordTypeCNAME, "owner").WithSetIdentifier("test-set-2"),
 		},
 	}
 	expected := &plan.Changes{
 		Create: []*endpoint.Endpoint{
 			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, ""),
+			newEndpointWithOwnerResource("multiple.test-zone.example.org", "lb3.loadbalancer.com", "", "owner", "ingress/default/my-ingress").WithSetIdentifier("test-set-3"),
+			newEndpointWithOwner("txt.multiple.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner,external-dns/resource=ingress/default/my-ingress\"", endpoint.RecordTypeTXT, "").WithSetIdentifier("test-set-3"),
 		},
 		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, ""),
+			newEndpointWithOwner("multiple.test-zone.example.org", "lb1.loadbalancer.com", endpoint.RecordTypeCNAME, "owner").WithSetIdentifier("test-set-1"),
+			newEndpointWithOwner("txt.multiple.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner\"", endpoint.RecordTypeTXT, "").WithSetIdentifier("test-set-1"),
 		},
 		UpdateNew: []*endpoint.Endpoint{
 			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, ""),
+			newEndpointWithOwnerResource("multiple.test-zone.example.org", "new.loadbalancer.com", endpoint.RecordTypeCNAME, "owner", "ingress/default/my-ingress-2").WithSetIdentifier("test-set-2"),
+			newEndpointWithOwner("txt.multiple.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner,external-dns/resource=ingress/default/my-ingress-2\"", endpoint.RecordTypeTXT, "").WithSetIdentifier("test-set-2"),
 		},
 		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, ""),
+			newEndpointWithOwner("multiple.test-zone.example.org", "lb2.loadbalancer.com", endpoint.RecordTypeCNAME, "owner").WithSetIdentifier("test-set-2"),
+			newEndpointWithOwner("txt.multiple.test-zone.example.org", "\"heritage=external-dns,external-dns/owner=owner\"", endpoint.RecordTypeTXT, "").WithSetIdentifier("test-set-2"),
 		},
 	}
 	p.OnApplyChanges = func(ctx context.Context, got *plan.Changes) {

source/dedup_source.go

@@ -43,7 +43,7 @@ func (ms *dedupSource) Endpoints() ([]*endpoint.Endpoint, error) {
 	}
 
 	for _, ep := range endpoints {
-		identifier := ep.DNSName + " / " + ep.Targets.String()
+		identifier := ep.DNSName + " / " + ep.SetIdentifier + " / " + ep.Targets.String()
 
 		if _, ok := collected[identifier]; ok {
 			log.Debugf("Removing duplicate endpoint %s", ep)

source/gateway.go

@@ -174,14 +174,14 @@ func (sc *gatewaySource) endpointsFromTemplate(config *istiomodel.Config) ([]*en
 		}
 	}
 
-	providerSpecific := getProviderSpecificAnnotations(config.Annotations)
+	providerSpecific, setIdentifier := getProviderSpecificAnnotations(config.Annotations)
 
 	var endpoints []*endpoint.Endpoint
 	// splits the FQDN template and removes the trailing periods
 	hostnameList := strings.Split(strings.Replace(hostnames, " ", "", -1), ",")
 	for _, hostname := range hostnameList {
 		hostname = strings.TrimSuffix(hostname, ".")
-		endpoints = append(endpoints, endpointsForHostname(hostname, targets, ttl, providerSpecific)...)
+		endpoints = append(endpoints, endpointsForHostname(hostname, targets, ttl, providerSpecific, setIdentifier)...)
 	}
 	return endpoints, nil
 }
@@ -266,7 +266,7 @@ func (sc *gatewaySource) endpointsFromGatewayConfig(config istiomodel.Config) ([
 
 	gateway := config.Spec.(*istionetworking.Gateway)
 
-	providerSpecific := getProviderSpecificAnnotations(config.Annotations)
+	providerSpecific, setIdentifier := getProviderSpecificAnnotations(config.Annotations)
 
 	for _, server := range gateway.Servers {
 		for _, host := range server.Hosts {
@@ -282,7 +282,7 @@ func (sc *gatewaySource) endpointsFromGatewayConfig(config istiomodel.Config) ([
 				host = parts[1]
 			}
 
-			endpoints = append(endpoints, endpointsForHostname(host, targets, ttl, providerSpecific)...)
+			endpoints = append(endpoints, endpointsForHostname(host, targets, ttl, providerSpecific, setIdentifier)...)
 		}
 	}
 
@@ -290,7 +290,7 @@ func (sc *gatewaySource) endpointsFromGatewayConfig(config istiomodel.Config) ([
 	if !sc.ignoreHostnameAnnotation {
 		hostnameList := getHostnamesFromAnnotations(config.Annotations)
 		for _, hostname := range hostnameList {
-			endpoints = append(endpoints, endpointsForHostname(hostname, targets, ttl, providerSpecific)...)
+			endpoints = append(endpoints, endpointsForHostname(hostname, targets, ttl, providerSpecific, setIdentifier)...)
 		}
 	}
 

source/ingress.go

@@ -187,14 +187,14 @@ func (sc *ingressSource) endpointsFromTemplate(ing *v1beta1.Ingress) ([]*endpoin
 		targets = targetsFromIngressStatus(ing.Status)
 	}
 
-	providerSpecific := getProviderSpecificAnnotations(ing.Annotations)
+	providerSpecific, setIdentifier := getProviderSpecificAnnotations(ing.Annotations)
 
 	var endpoints []*endpoint.Endpoint
 	// splits the FQDN template and removes the trailing periods
 	hostnameList := strings.Split(strings.Replace(hostnames, " ", "", -1), ",")
 	for _, hostname := range hostnameList {
 		hostname = strings.TrimSuffix(hostname, ".")
-		endpoints = append(endpoints, endpointsForHostname(hostname, targets, ttl, providerSpecific)...)
+		endpoints = append(endpoints, endpointsForHostname(hostname, targets, ttl, providerSpecific, setIdentifier)...)
 	}
 	return endpoints, nil
 }
@@ -261,13 +261,13 @@ func endpointsFromIngress(ing *v1beta1.Ingress, ignoreHostnameAnnotation bool) [
 		targets = targetsFromIngressStatus(ing.Status)
 	}
 
-	providerSpecific := getProviderSpecificAnnotations(ing.Annotations)
+	providerSpecific, setIdentifier := getProviderSpecificAnnotations(ing.Annotations)
 
 	for _, rule := range ing.Spec.Rules {
 		if rule.Host == "" {
 			continue
 		}
-		endpoints = append(endpoints, endpointsForHostname(rule.Host, targets, ttl, providerSpecific)...)
+		endpoints = append(endpoints, endpointsForHostname(rule.Host, targets, ttl, providerSpecific, setIdentifier)...)
 	}
 
 	for _, tls := range ing.Spec.TLS {
@@ -275,7 +275,7 @@ func endpointsFromIngress(ing *v1beta1.Ingress, ignoreHostnameAnnotation bool) [
 			if host == "" {
 				continue
 			}
-			endpoints = append(endpoints, endpointsForHostname(host, targets, ttl, providerSpecific)...)
+			endpoints = append(endpoints, endpointsForHostname(host, targets, ttl, providerSpecific, setIdentifier)...)
 		}
 	}
 
@@ -283,7 +283,7 @@ func endpointsFromIngress(ing *v1beta1.Ingress, ignoreHostnameAnnotation bool) [
 	if !ignoreHostnameAnnotation {
 		hostnameList := getHostnamesFromAnnotations(ing.Annotations)
 		for _, hostname := range hostnameList {
-			endpoints = append(endpoints, endpointsForHostname(hostname, targets, ttl, providerSpecific)...)
+			endpoints = append(endpoints, endpointsForHostname(hostname, targets, ttl, providerSpecific, setIdentifier)...)
 		}
 	}
 	return endpoints

source/ingressroute.go

@@ -209,14 +209,14 @@ func (sc *ingressRouteSource) endpointsFromTemplate(ingressRoute *contourapi.Ing
 		}
 	}
 
-	providerSpecific := getProviderSpecificAnnotations(ingressRoute.Annotations)
+	providerSpecific, setIdentifier := getProviderSpecificAnnotations(ingressRoute.Annotations)
 
 	var endpoints []*endpoint.Endpoint
 	// splits the FQDN template and removes the trailing periods
 	hostnameList := strings.Split(strings.Replace(hostnames, " ", "", -1), ",")
 	for _, hostname := range hostnameList {
 		hostname = strings.TrimSuffix(hostname, ".")
-		endpoints = append(endpoints, endpointsForHostname(hostname, targets, ttl, providerSpecific)...)
+		endpoints = append(endpoints, endpointsForHostname(hostname, targets, ttl, providerSpecific, setIdentifier)...)
 	}
 	return endpoints, nil
 }
@@ -302,11 +302,11 @@ func (sc *ingressRouteSource) endpointsFromIngressRoute(ingressRoute *contourapi
 		}
 	}
 
-	providerSpecific := getProviderSpecificAnnotations(ingressRoute.Annotations)
+	providerSpecific, setIdentifier := getProviderSpecificAnnotations(ingressRoute.Annotations)
 
 	if virtualHost := ingressRoute.Spec.VirtualHost; virtualHost != nil {
 		if fqdn := virtualHost.Fqdn; fqdn != "" {
-			endpoints = append(endpoints, endpointsForHostname(fqdn, targets, ttl, providerSpecific)...)
+			endpoints = append(endpoints, endpointsForHostname(fqdn, targets, ttl, providerSpecific, setIdentifier)...)
 		}
 	}
 
@@ -314,7 +314,7 @@ func (sc *ingressRouteSource) endpointsFromIngressRoute(ingressRoute *contourapi
 	if !sc.ignoreHostnameAnnotation {
 		hostnameList := getHostnamesFromAnnotations(ingressRoute.Annotations)
 		for _, hostname := range hostnameList {
-			endpoints = append(endpoints, endpointsForHostname(hostname, targets, ttl, providerSpecific)...)
+			endpoints = append(endpoints, endpointsForHostname(hostname, targets, ttl, providerSpecific, setIdentifier)...)
 		}
 	}
 

source/service.go

@@ -285,10 +285,10 @@ func (sc *serviceSource) endpointsFromTemplate(svc *v1.Service) ([]*endpoint.End
 		return nil, fmt.Errorf("failed to apply template on service %s: %v", svc.String(), err)
 	}
 
-	providerSpecific := getProviderSpecificAnnotations(svc.Annotations)
+	providerSpecific, setIdentifier := getProviderSpecificAnnotations(svc.Annotations)
 	hostnameList := strings.Split(strings.Replace(buf.String(), " ", "", -1), ",")
 	for _, hostname := range hostnameList {
-		endpoints = append(endpoints, sc.generateEndpoints(svc, hostname, providerSpecific)...)
+		endpoints = append(endpoints, sc.generateEndpoints(svc, hostname, providerSpecific, setIdentifier)...)
 	}
 
 	return endpoints, nil
@@ -299,10 +299,10 @@ func (sc *serviceSource) endpoints(svc *v1.Service) []*endpoint.Endpoint {
 	var endpoints []*endpoint.Endpoint
 	// Skip endpoints if we do not want entries from annotations
 	if !sc.ignoreHostnameAnnotation {
-		providerSpecific := getProviderSpecificAnnotations(svc.Annotations)
+		providerSpecific, setIdentifier := getProviderSpecificAnnotations(svc.Annotations)
 		hostnameList := getHostnamesFromAnnotations(svc.Annotations)
 		for _, hostname := range hostnameList {
-			endpoints = append(endpoints, sc.generateEndpoints(svc, hostname, providerSpecific)...)
+			endpoints = append(endpoints, sc.generateEndpoints(svc, hostname, providerSpecific, setIdentifier)...)
 		}
 	}
 	return endpoints
@@ -358,7 +358,7 @@ func (sc *serviceSource) setResourceLabel(service *v1.Service, endpoints []*endp
 	}
 }
 
-func (sc *serviceSource) generateEndpoints(svc *v1.Service, hostname string, providerSpecific endpoint.ProviderSpecific) []*endpoint.Endpoint {
+func (sc *serviceSource) generateEndpoints(svc *v1.Service, hostname string, providerSpecific endpoint.ProviderSpecific, setIdentifier string) []*endpoint.Endpoint {
 	hostname = strings.TrimSuffix(hostname, ".")
 	ttl, err := getTTLFromAnnotations(svc.Annotations)
 	if err != nil {
@@ -366,21 +366,19 @@ func (sc *serviceSource) generateEndpoints(svc *v1.Service, hostname string, pro
 	}
 
 	epA := &endpoint.Endpoint{
-		RecordTTL:        ttl,
+		RecordTTL:  ttl,
-		RecordType:       endpoint.RecordTypeA,
+		RecordType: endpoint.RecordTypeA,
-		Labels:           endpoint.NewLabels(),
+		Labels:     endpoint.NewLabels(),
-		Targets:          make(endpoint.Targets, 0, defaultTargetsCapacity),
+		Targets:    make(endpoint.Targets, 0, defaultTargetsCapacity),
-		DNSName:          hostname,
+		DNSName:    hostname,
-		ProviderSpecific: providerSpecific,
 	}
 
 	epCNAME := &endpoint.Endpoint{
-		RecordTTL:        ttl,
+		RecordTTL:  ttl,
-		RecordType:       endpoint.RecordTypeCNAME,
+		RecordType: endpoint.RecordTypeCNAME,
-		Labels:           endpoint.NewLabels(),
+		Labels:     endpoint.NewLabels(),
-		Targets:          make(endpoint.Targets, 0, defaultTargetsCapacity),
+		Targets:    make(endpoint.Targets, 0, defaultTargetsCapacity),
-		DNSName:          hostname,
+		DNSName:    hostname,
-		ProviderSpecific: providerSpecific,
 	}
 
 	var endpoints []*endpoint.Endpoint
@@ -423,6 +421,10 @@ func (sc *serviceSource) generateEndpoints(svc *v1.Service, hostname string, pro
 	if len(epCNAME.Targets) > 0 {
 		endpoints = append(endpoints, epCNAME)
 	}
+	for _, endpoint := range endpoints {
+		endpoint.ProviderSpecific = providerSpecific
+		endpoint.SetIdentifier = setIdentifier
+	}
 	return endpoints
 }
 

source/source.go

@@ -45,6 +45,8 @@ const (
 const (
 	// The annotation used for determining if traffic will go through Cloudflare
 	CloudflareProxiedKey = "external-dns.alpha.kubernetes.io/cloudflare-proxied"
+
+	SetIdentifierKey = "external-dns.alpha.kubernetes.io/set-identifier"
 )
 
 const (
@@ -86,7 +88,7 @@ func getAliasFromAnnotations(annotations map[string]string) bool {
 	return exists && aliasAnnotation == "true"
 }
 
-func getProviderSpecificAnnotations(annotations map[string]string) endpoint.ProviderSpecific {
+func getProviderSpecificAnnotations(annotations map[string]string) (endpoint.ProviderSpecific, string) {
 	providerSpecificAnnotations := endpoint.ProviderSpecific{}
 
 	v, exists := annotations[CloudflareProxiedKey]
@@ -102,7 +104,19 @@ func getProviderSpecificAnnotations(annotations map[string]string) endpoint.Prov
 			Value: "true",
 		})
 	}
-	return providerSpecificAnnotations
+	setIdentifier := ""
+	for k, v := range annotations {
+		if k == SetIdentifierKey {
+			setIdentifier = v
+		} else if strings.HasPrefix(k, "external-dns.alpha.kubernetes.io/aws-") {
+			attr := strings.TrimPrefix(k, "external-dns.alpha.kubernetes.io/aws-")
+			providerSpecificAnnotations = append(providerSpecificAnnotations, endpoint.ProviderSpecificProperty{
+				Name:  fmt.Sprintf("aws/%s", attr),
+				Value: v,
+			})
+		}
+	}
+	return providerSpecificAnnotations, setIdentifier
 }
 
 // getTargetsFromTargetAnnotation gets endpoints from optional "target" annotation.
@@ -133,7 +147,7 @@ func suitableType(target string) string {
 }
 
 // endpointsForHostname returns the endpoint objects for each host-target combination.
-func endpointsForHostname(hostname string, targets endpoint.Targets, ttl endpoint.TTL, providerSpecific endpoint.ProviderSpecific) []*endpoint.Endpoint {
+func endpointsForHostname(hostname string, targets endpoint.Targets, ttl endpoint.TTL, providerSpecific endpoint.ProviderSpecific, setIdentifier string) []*endpoint.Endpoint {
 	var endpoints []*endpoint.Endpoint
 
 	var aTargets endpoint.Targets
@@ -156,6 +170,7 @@ func endpointsForHostname(hostname string, targets endpoint.Targets, ttl endpoin
 			RecordType:       endpoint.RecordTypeA,
 			Labels:           endpoint.NewLabels(),
 			ProviderSpecific: providerSpecific,
+			SetIdentifier:    setIdentifier,
 		}
 		endpoints = append(endpoints, epA)
 	}
@@ -168,6 +183,7 @@ func endpointsForHostname(hostname string, targets endpoint.Targets, ttl endpoin
 			RecordType:       endpoint.RecordTypeCNAME,
 			Labels:           endpoint.NewLabels(),
 			ProviderSpecific: providerSpecific,
+			SetIdentifier:    setIdentifier,
 		}
 		endpoints = append(endpoints, epCNAME)
 	}