TXTRegistry: ownerID tracking (for multi-target records) via internal…

… state rather than Endpoint.Labels

plan/plan.go

@@ -73,7 +73,7 @@ func (p *Plan) Calculate() *Plan {
 		}
 
 		changes.UpdateOld = append(changes.UpdateOld, current)
-		desired.MergeLabels(current.Labels)     // inherit the labels from the dns provider, including Owner ID
+		desired.MergeLabels(current.Labels)     // inherit the labels from the dns provider
 		desired.RecordType = current.RecordType // inherit the type from the dns provider
 
 		changes.UpdateNew = append(changes.UpdateNew, desired)

registry/registry.go

@@ -19,7 +19,6 @@ package registry
 import (
 	"github.com/kubernetes-incubator/external-dns/endpoint"
 	"github.com/kubernetes-incubator/external-dns/plan"
-	log "github.com/sirupsen/logrus"
 )
 
 // Registry is an interface which should enables ownership concept in external-dns
@@ -30,16 +29,3 @@ type Registry interface {
 	Records() ([]*endpoint.Endpoint, error)
 	ApplyChanges(changes *plan.Changes) error
 }
-
-//TODO(ideahitme): consider moving this to Plan
-func filterOwnedRecords(ownerID string, eps []*endpoint.Endpoint) []*endpoint.Endpoint {
-	filtered := []*endpoint.Endpoint{}
-	for _, ep := range eps {
-		if endpointOwner, ok := ep.Labels[endpoint.OwnerLabelKey]; !ok || endpointOwner != ownerID {
-			log.Debugf(`Skipping endpoint %v because owner id does not match, found: "%s", required: "%s"`, ep, endpointOwner, ownerID)
-			continue
-		}
-		filtered = append(filtered, ep)
-	}
-	return filtered
-}

registry/txt.go

@@ -20,6 +20,7 @@ import (
 	"errors"
 
 	"fmt"
+	log "github.com/sirupsen/logrus"
 	"regexp"
 	"strings"
 
@@ -38,6 +39,8 @@ type TXTRegistry struct {
 	provider provider.Provider
 	ownerID  string //refers to the owner id of the current instance
 	mapper   nameMapper
+	ownerMap map[string]string // map DNSName => ownerID
+	recCount map[string]int    // map DNSName => number of records
 }
 
 // NewTXTRegistry returns new TXTRegistry object
@@ -66,64 +69,96 @@ func (im *TXTRegistry) Records() ([]*endpoint.Endpoint, error) {
 
 	endpoints := make([]*endpoint.Endpoint, 0)
 
-	ownerMap := map[string]string{}
+	im.ownerMap = make(map[string]string)
+	im.recCount = make(map[string]int)
 
 	for _, record := range records {
 		if record.RecordType != endpoint.RecordTypeTXT {
 			endpoints = append(endpoints, record)
+			im.recCount[record.DNSName]++
 			continue
 		}
+
 		ownerID := im.extractOwnerID(record.Target)
 		if ownerID == "" {
 			//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
 		}
+
 		endpointDNSName := im.mapper.toEndpointName(record.DNSName)
-		ownerMap[endpointDNSName] = ownerID
+		im.ownerMap[endpointDNSName] = ownerID
 	}
 
-	for _, ep := range endpoints {
-		ep.Labels[endpoint.OwnerLabelKey] = ownerMap[ep.DNSName]
-	}
+	log.Debugf("after scanning: ownerMap: %s, recCount: %s", im.ownerMap, im.recCount)
 
 	return endpoints, err
 }
 
 // ApplyChanges updates dns provider with the changes
 // for each created/deleted record it will also take into account TXT records for creation/deletion
 func (im *TXTRegistry) ApplyChanges(changes *plan.Changes) error {
+	if im.ownerMap == nil {
+		// Records() never called yet or error during last ApplyChanges() call => rescan
+		_, err := im.Records()
+		if err != nil {
+			return err
+		}
+	}
+
 	filteredChanges := &plan.Changes{
-		Create:    changes.Create,
-		UpdateNew: filterOwnedRecords(im.ownerID, changes.UpdateNew),
-		UpdateOld: filterOwnedRecords(im.ownerID, changes.UpdateOld),
-		Delete:    filterOwnedRecords(im.ownerID, changes.Delete),
+		Create:    []*endpoint.Endpoint{},
+		UpdateNew: im.filterOwnedRecords(changes.UpdateNew),
+		UpdateOld: im.filterOwnedRecords(changes.UpdateOld),
+		Delete:    im.filterOwnedRecords(changes.Delete),
 	}
 
-	// create one create / delete TXT record per created/deleted DNSName
+	for _, r := range changes.Create {
+		rOwner := im.ownerMap[r.DNSName]
 
-	createdDomains := map[string]bool{}
+		if rOwner == "" || rOwner == im.ownerID {
+			filteredChanges.Create = append(filteredChanges.Create, r)
 
-	for _, r := range filteredChanges.Create {
-		if !createdDomains[r.DNSName] {
-			txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), im.getTXTLabel(), endpoint.RecordTypeTXT)
-			filteredChanges.Create = append(filteredChanges.Create, txt)
-			createdDomains[r.DNSName] = true
+			im.ownerMap[r.DNSName] = im.ownerID
+			im.recCount[r.DNSName]++
+
+			if 1 == im.recCount[r.DNSName] {
+				txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), im.getTXTLabel(), endpoint.RecordTypeTXT)
+				filteredChanges.Create = append(filteredChanges.Create, txt)
+			}
 		}
 	}
 
-	deletedDomains := map[string]bool{}
-
 	for _, r := range filteredChanges.Delete {
-		if !deletedDomains[r.DNSName] {
+		if im.recCount[r.DNSName]--; im.recCount[r.DNSName] <= 0 {
+			im.recCount[r.DNSName] = 0
+			delete(im.ownerMap, r.DNSName)
 			txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), im.getTXTLabel(), endpoint.RecordTypeTXT)
 			filteredChanges.Delete = append(filteredChanges.Delete, txt)
-			deletedDomains[r.DNSName] = true
 		}
 	}
 
-	return im.provider.ApplyChanges(filteredChanges)
+	log.Debugf("before provider.ApplyChanges: ownerMap: %s, recCount: %s", im.ownerMap, im.recCount)
+
+	err := im.provider.ApplyChanges(filteredChanges)
+	if err != nil {
+		// error occured in the provider => we don't know which records were stored => force re-scanning on the next call
+		im.ownerMap = nil
+	}
+	return err
+}
+
+func (im *TXTRegistry) filterOwnedRecords(eps []*endpoint.Endpoint) []*endpoint.Endpoint {
+	filtered := []*endpoint.Endpoint{}
+	for _, ep := range eps {
+		if endpointOwner, ok := im.ownerMap[ep.DNSName]; !ok || endpointOwner != im.ownerID {
+			log.Debugf(`Skipping endpoint %v because owner id does not match, found: "%s", required: "%s"`, ep, endpointOwner, im.ownerID)
+			continue
+		}
+		filtered = append(filtered, ep)
+	}
+	return filtered
 }
 
 /**