Fix ownership tracking with noop registry

Author: Jonas Weber

cmd/webhook/main.go

@@ -1,25 +1,30 @@
 package main
 
 import (
+	"external-dns-openstack-webhook/internal/designate/provider"
+	"external-dns-openstack-webhook/internal/metrics"
+	"flag"
 	"net"
 	"net/http"
 
 	log "github.com/sirupsen/logrus"
 
-	"external-dns-openstack-webhook/internal/designate/provider"
-	"external-dns-openstack-webhook/internal/metrics"
-
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"sigs.k8s.io/external-dns/endpoint"
 	"sigs.k8s.io/external-dns/provider/webhook/api"
 )
 
-const (
-	webhookServerAddr = "127.0.0.1:8888"
-	statusServerAddr  = "0.0.0.0:8080"
-)
-
 func main() {
+	var webhookServerAddr string
+	var statusServerAddr string
+	var ownerID string
+	var dryRun bool
+	flag.StringVar(&webhookServerAddr, "webhook-address", "127.0.0.1:8888", "Address for webhook to listen on")
+	flag.StringVar(&statusServerAddr, "status-address", "0.0.0.0:8080", "Address for status server to listen on")
+	flag.StringVar(&ownerID, "owner-id", "development", "Owner ID to apply instead of TXT registry.")
+	flag.BoolVar(&dryRun, "dry-run", false, "Do not actually change anything")
+	flag.Parse()
+
 	log.SetLevel(log.DebugLevel)
 
 	startedChan := make(chan struct{})
@@ -56,7 +61,7 @@ func main() {
 	}()
 
 	epf := endpoint.NewDomainFilter([]string{})
-	dp, err := provider.NewDesignateProvider(*epf, false)
+	dp, err := provider.NewDesignateProvider(*epf, dryRun, ownerID)
 	if err != nil {
 		log.Fatalf("NewDesignateProvider: %v", err)
 		metrics.OpenstackConnectionMetric.Set(0)

internal/designate/provider/provider.go

@@ -19,6 +19,7 @@ package provider
 
 import (
 	"context"
+	"errors"
 	"external-dns-openstack-webhook/internal/designate/client"
 	"fmt"
 	"strings"
@@ -52,10 +53,12 @@ type designateProvider struct {
 	// only consider hosted zones managing domains ending in this suffix
 	domainFilter endpoint.DomainFilter
 	dryRun       bool
+
+	ownerID string
 }
 
 // NewDesignateProvider is a factory function for OpenStack designate providers
-func NewDesignateProvider(domainFilter endpoint.DomainFilter, dryRun bool) (provider.Provider, error) {
+func NewDesignateProvider(domainFilter endpoint.DomainFilter, dryRun bool, ownerID string) (provider.Provider, error) {
 	client, err := client.NewDesignateClient()
 	if err != nil {
 		return nil, err
@@ -64,6 +67,7 @@ func NewDesignateProvider(domainFilter endpoint.DomainFilter, dryRun bool) (prov
 		client:       client,
 		domainFilter: domainFilter,
 		dryRun:       dryRun,
+		ownerID:      ownerID,
 	}, nil
 }
 
@@ -156,10 +160,25 @@ func (p designateProvider) Records(ctx context.Context) ([]*endpoint.Endpoint, e
 				}
 
 				ep := endpoint.NewEndpointWithTTL(recordSet.Name, recordSet.Type, endpoint.TTL(recordSet.TTL), recordSet.Records...)
-				ep.Labels[designateRecordSetID] = recordSet.ID
-				ep.Labels[designateZoneID] = recordSet.ZoneID
-				ep.Labels[designateOriginalRecords] = strings.Join(recordSet.Records, "\000")
-				result = append(result, ep)
+
+				parsedLabels, err := endpoint.NewLabelsFromStringPlain(recordSet.Description)
+				if err != nil && !errors.Is(err, endpoint.ErrInvalidHeritage) {
+					return fmt.Errorf("unable to parse label: %w", err)
+				}
+				for k, v := range parsedLabels {
+					ep.WithLabel(k, v)
+				}
+
+				ep.
+					WithLabel(designateRecordSetID, recordSet.ID).
+					WithLabel(designateZoneID, recordSet.ZoneID).
+					WithLabel(designateOriginalRecords, strings.Join(recordSet.Records, "\000"))
+
+				if ep.IsOwnedBy(p.ownerID) {
+					result = append(result, ep)
+				} else {
+					log.Debugf("SKIP %s/%s due to owner not matching %s", recordSet.Name, recordSet.Type, p.ownerID)
+				}
 
 				return nil
 			},
@@ -174,7 +193,7 @@ func (p designateProvider) Records(ctx context.Context) ([]*endpoint.Endpoint, e
 
 func (p designateProvider) supportedRecordType(recordType string) bool {
 	switch recordType {
-	case endpoint.RecordTypeA, endpoint.RecordTypeTXT, endpoint.RecordTypeCNAME, endpoint.RecordTypeNS, endpoint.RecordTypeMX:
+	case endpoint.RecordTypeA, endpoint.RecordTypeTXT, endpoint.RecordTypeCNAME, endpoint.RecordTypeNS, endpoint.RecordTypeMX, "CAA":
 		return true
 	default:
 		return false
@@ -189,70 +208,63 @@ type recordSet struct {
 	recordSetID string
 	ttl         int
 	names       map[string]bool
+	description string
 }
 
 // adds endpoint into recordset aggregation, loading original values from endpoint labels first
 func addEndpoint(ep *endpoint.Endpoint, recordSets map[string]*recordSet, oldEndpoints []*endpoint.Endpoint, delete bool) {
 	key := fmt.Sprintf("%s/%s", ep.DNSName, ep.RecordType)
+
 	rs := recordSets[key]
 	if rs == nil {
+		recordedLabels := endpoint.NewLabels()
+
+		// Filter out our internal labels
+		for k, v := range ep.Labels {
+			switch k {
+			case designateRecordSetID, designateZoneID, designateOriginalRecords:
+			default:
+				recordedLabels[k] = v
+			}
+		}
 		rs = &recordSet{
-			dnsName:    canonicalizeDomainName(ep.DNSName),
-			recordType: ep.RecordType,
-			names:      make(map[string]bool),
+			dnsName:     canonicalizeDomainName(ep.DNSName),
+			recordType:  ep.RecordType,
+			names:       make(map[string]bool),
+			description: recordedLabels.SerializePlain(false),
 		}
 	}
 
-	addDesignateIDLabelsFromExistingEndpoints(oldEndpoints, ep)
-
-	if rs.zoneID == "" {
-		rs.zoneID = ep.Labels[designateZoneID]
+	if zoneID, ok := ep.Labels[designateZoneID]; ok {
+		rs.zoneID = zoneID
 	}
-	if rs.recordSetID == "" {
-		rs.recordSetID = ep.Labels[designateRecordSetID]
+	if recordSetID, ok := ep.Labels[designateRecordSetID]; ok {
+		rs.recordSetID = recordSetID
 	}
-	rs.ttl = int(ep.RecordTTL)
-	for _, rec := range strings.Split(ep.Labels[designateOriginalRecords], "\000") {
-		if _, ok := rs.names[rec]; !ok && rec != "" {
-			rs.names[rec] = true
+	if origRecords, ok := ep.Labels[designateOriginalRecords]; ok {
+		for _, rec := range strings.Split(origRecords, "\000") {
+			if _, ok := rs.names[rec]; !ok && rec != "" {
+				rs.names[rec] = true
+			}
 		}
 	}
+	rs.ttl = int(ep.RecordTTL)
+
 	targets := ep.Targets
 	if ep.RecordType == endpoint.RecordTypeCNAME || ep.RecordType == endpoint.RecordTypeNS {
 		targets = canonicalizeDomainNames(targets)
 	}
+
 	if ep.RecordType == endpoint.RecordTypeMX {
 		targets = canonicalizeDomainNamesForMX(targets)
 	}
+
 	for _, t := range targets {
 		rs.names[t] = !delete
 	}
 	recordSets[key] = rs
 }
 
-// addDesignateIDLabelsFromExistingEndpoints adds the labels identified by the constants designateZoneID and designateRecordSetID
-// to an Endpoint. Therefore, it searches all given existing endpoints for an endpoint with the same record type and record
-// value. If the given Endpoint already has the labels set, they are left untouched. This fixes an issue with the
-// TXTRegistry which generates new TXT entries instead of updating the old ones.
-func addDesignateIDLabelsFromExistingEndpoints(existingEndpoints []*endpoint.Endpoint, ep *endpoint.Endpoint) {
-	_, hasZoneIDLabel := ep.Labels[designateZoneID]
-	_, hasRecordSetIDLabel := ep.Labels[designateRecordSetID]
-	if hasZoneIDLabel && hasRecordSetIDLabel {
-		return
-	}
-	for _, oep := range existingEndpoints {
-		if ep.RecordType == oep.RecordType && ep.DNSName == oep.DNSName {
-			if !hasZoneIDLabel {
-				ep.Labels[designateZoneID] = oep.Labels[designateZoneID]
-			}
-			if !hasRecordSetIDLabel {
-				ep.Labels[designateRecordSetID] = oep.Labels[designateRecordSetID]
-			}
-			return
-		}
-	}
-}
-
 // ApplyChanges applies a given set of changes in a given zone.
 func (p designateProvider) ApplyChanges(ctx context.Context, changes *plan.Changes) error {
 	managedZones, err := p.getZones(ctx)
@@ -265,17 +277,25 @@ func (p designateProvider) ApplyChanges(ctx context.Context, changes *plan.Chang
 		return fmt.Errorf("failed to fetch active records: %w", err)
 	}
 
+	for _, ep := range changes.Create {
+		log.Debugf("CREATE %v", ep)
+	}
+
 	recordSets := map[string]*recordSet{}
+
+	logPlan(changes)
+
 	for _, ep := range changes.Create {
+		ep.Labels[endpoint.OwnerLabelKey] = p.ownerID
 		addEndpoint(ep, recordSets, endpoints, false)
 	}
-	for _, ep := range changes.UpdateOld {
+	for _, ep := range endpoint.FilterEndpointsByOwnerID(p.ownerID, changes.UpdateOld) {
 		addEndpoint(ep, recordSets, endpoints, true)
 	}
-	for _, ep := range changes.UpdateNew {
+	for _, ep := range endpoint.FilterEndpointsByOwnerID(p.ownerID, changes.UpdateNew) {
 		addEndpoint(ep, recordSets, endpoints, false)
 	}
-	for _, ep := range changes.Delete {
+	for _, ep := range endpoint.FilterEndpointsByOwnerID(p.ownerID, changes.Delete) {
 		addEndpoint(ep, recordSets, endpoints, true)
 	}
 
@@ -287,6 +307,24 @@ func (p designateProvider) ApplyChanges(ctx context.Context, changes *plan.Chang
 	return err
 }
 
+func logPlan(plan *plan.Changes) {
+	infos := []struct {
+		name string
+		list []*endpoint.Endpoint
+	}{
+		{"CREATE", plan.Create},
+		{"UPDOLD", plan.UpdateOld},
+		{"UPDNEW", plan.UpdateNew},
+		{"DELETE", plan.Delete},
+	}
+
+	for _, i := range infos {
+		for _, ep := range i.list {
+			log.Debugf("%s %v", i.name, *ep)
+		}
+	}
+}
+
 // apply recordset changes by inserting/updating/deleting recordsets
 func (p designateProvider) upsertRecordSet(ctx context.Context, rs *recordSet, managedZones map[string]string) error {
 	if rs.zoneID == "" {
@@ -307,10 +345,11 @@ func (p designateProvider) upsertRecordSet(ctx context.Context, rs *recordSet, m
 	}
 	if rs.recordSetID == "" {
 		opts := recordsets.CreateOpts{
-			Name:    rs.dnsName,
-			Type:    rs.recordType,
-			Records: records,
-			TTL:     rs.ttl,
+			Name:        rs.dnsName,
+			Type:        rs.recordType,
+			Records:     records,
+			TTL:         rs.ttl,
+			Description: rs.description,
 		}
 		log.Infof("Creating records: %s/%s: %s", rs.dnsName, rs.recordType, strings.Join(records, ","))
 		if p.dryRun {
@@ -326,8 +365,9 @@ func (p designateProvider) upsertRecordSet(ctx context.Context, rs *recordSet, m
 		return p.client.DeleteRecordSet(ctx, rs.zoneID, rs.recordSetID)
 	} else {
 		opts := recordsets.UpdateOpts{
-			Records: records,
-			TTL:     &rs.ttl,
+			Records:     records,
+			TTL:         &rs.ttl,
+			Description: &rs.description,
 		}
 		log.Infof("Updating records: %s/%s: %s", rs.dnsName, rs.recordType, strings.Join(records, ","))
 		if p.dryRun {

internal/designate/provider/provider_test.go

@@ -206,7 +206,7 @@ clouds:
 	os.Setenv("OS_CLOUD", "unittest")
 	os.Setenv("OS_CACERT", tmpfile.Name())
 
-	if _, err := NewDesignateProvider(endpoint.DomainFilter{}, true); err != nil {
+	if _, err := NewDesignateProvider(endpoint.DomainFilter{}, true, "dev"); err != nil {
 		t.Fatalf("Failed to initialize Designate provider: %s", err)
 	}
 }