Fix planning for multi-cluster dual stack record types

When AAAA multi-target / dual stack support was
added via #2461 it broke ownership of domains across
different clusters with different ingress records types.

For example if 2 clusters manage the same zone,
1 cluster uses A records and the other uses CNAME
records, when each record type is treated as a separate
planning record, it will cause ownership to bounce back
and forth and records to be constantly created and
deleted.

This change updates the planner to keep track of multiple
current records for a domain. This allows for A and AAAA
records to exist for a domain while allowing record type
changes.

The planner will ignore desired records for a domain that
represent conflicting record types allowed by RFC 1034 3.6.2.
For example if the desired records for a domain contains
a CNAME record plus any other record type no changes for
that domain will be planned.

The planner now contains an owned record filter provided
by the registry. This allows the planner to accurately plan
create updates when there are record type changes between
the current and desired endpoints. Without this filter the
planner could add create changes for domains not owned
by the controller.

controller/controller.go

@@ -223,6 +223,7 @@ func (c *Controller) RunOnce(ctx context.Context) error {
 		DomainFilter:       endpoint.MatchAllDomainFilters{c.DomainFilter, c.Registry.GetDomainFilter()},
 		PropertyComparator: c.Registry.PropertyValuesEqual,
 		ManagedRecords:     c.ManagedRecordTypes,
+		OwnedRecordFilter:  c.Registry.GetOwnedRecordFilter(),
 	}
 
 	plan = plan.Calculate()

endpoint/endpoint.go

@@ -281,6 +281,44 @@ func (e *Endpoint) String() string {
 	return fmt.Sprintf("%s %d IN %s %s %s %s", e.DNSName, e.RecordTTL, e.RecordType, e.SetIdentifier, e.Targets, e.ProviderSpecific)
 }
 
+// EndpointFilterInterface matches endpoints
+type EndpointFilterInterface interface {
+	// Match returns true if the endpoint matches the filter, false otherwise
+	Match(ep *Endpoint) bool
+}
+
+// Apply filter to slice of endpoints and return new filtered slice that includes
+// only endpoints that match.
+func ApplyEndpointFilter(filter EndpointFilterInterface, eps []*Endpoint) []*Endpoint {
+	filtered := []*Endpoint{}
+	for _, ep := range eps {
+		if filter.Match(ep) {
+			filtered = append(filtered, ep)
+		}
+	}
+
+	return filtered
+}
+
+// NewOwnedRecordFilter returns endpoint filter that matches records with a owner
+// label that matches the given owner id.
+func NewOwnedRecordFilter(ownerID string) EndpointFilterInterface {
+	return ownedRecordFilter{ownerID: ownerID}
+}
+
+type ownedRecordFilter struct {
+	ownerID string
+}
+
+func (f ownedRecordFilter) Match(ep *Endpoint) bool {
+	if endpointOwner, ok := ep.Labels[OwnerLabelKey]; !ok || endpointOwner != f.ownerID {
+		log.Debugf(`Skipping endpoint %v because owner id does not match, found: "%s", required: "%s"`, ep, endpointOwner, f.ownerID)
+		return false
+	} else {
+		return true
+	}
+}
+
 // DNSEndpointSpec defines the desired state of DNSEndpoint
 type DNSEndpointSpec struct {
 	Endpoints []*Endpoint `json:"endpoints,omitempty"`

endpoint/endpoint_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package endpoint
 
 import (
+	"reflect"
 	"testing"
 )
 
@@ -115,3 +116,128 @@ func TestIsLess(t *testing.T) {
 		}
 	}
 }
+
+func TestOwnedRecordFilterMatch(t *testing.T) {
+	type fields struct {
+		ownerID string
+	}
+	type args struct {
+		ep *Endpoint
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+		want   bool
+	}{
+		{
+			name:   "no labels",
+			fields: fields{ownerID: "foo"},
+			args: args{ep: &Endpoint{
+				DNSName:    "foo.com",
+				RecordType: RecordTypeA,
+			}},
+			want: false,
+		},
+		{
+			name:   "no owner label",
+			fields: fields{ownerID: "foo"},
+			args: args{ep: &Endpoint{
+				DNSName:    "foo.com",
+				RecordType: RecordTypeA,
+				Labels:     NewLabels(),
+			}},
+			want: false,
+		},
+		{
+			name:   "owner not matched",
+			fields: fields{ownerID: "foo"},
+			args: args{ep: &Endpoint{
+				DNSName:    "foo.com",
+				RecordType: RecordTypeA,
+				Labels: Labels{
+					OwnerLabelKey: "bar",
+				},
+			}},
+			want: false,
+		},
+		{
+			name:   "owner matched",
+			fields: fields{ownerID: "foo"},
+			args: args{ep: &Endpoint{
+				DNSName:    "foo.com",
+				RecordType: RecordTypeA,
+				Labels: Labels{
+					OwnerLabelKey: "foo",
+				},
+			}},
+			want: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			f := ownedRecordFilter{
+				ownerID: tt.fields.ownerID,
+			}
+			if got := f.Match(tt.args.ep); got != tt.want {
+				t.Errorf("ownedRecordFilter.Match() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestApplyEndpointFilter(t *testing.T) {
+	foo1 := &Endpoint{
+		DNSName:    "foo.com",
+		RecordType: RecordTypeA,
+		Labels: Labels{
+			OwnerLabelKey: "foo",
+		},
+	}
+	foo2 := &Endpoint{
+		DNSName:    "foo.com",
+		RecordType: RecordTypeCNAME,
+		Labels: Labels{
+			OwnerLabelKey: "foo",
+		},
+	}
+	bar := &Endpoint{
+		DNSName:    "foo.com",
+		RecordType: RecordTypeA,
+		Labels: Labels{
+			OwnerLabelKey: "bar",
+		},
+	}
+	type args struct {
+		filter EndpointFilterInterface
+		eps    []*Endpoint
+	}
+	tests := []struct {
+		name string
+		args args
+		want []*Endpoint
+	}{
+		{
+			name: "filter values",
+			args: args{
+				filter: NewOwnedRecordFilter("foo"),
+				eps: []*Endpoint{
+					foo1,
+					foo2,
+					bar,
+				},
+			},
+			want: []*Endpoint{
+				foo1,
+				foo2,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := ApplyEndpointFilter(tt.args.filter, tt.args.eps); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("ApplyEndpointFilter() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}