Merge pull request #44626 from madhusudancs/fed-dns-paged-list

Automatic merge from submit-queue (batch tested with PRs 44626, 45641)

Update Google Cloud DNS provider Rrset.Get(name) method to return a list and change the `Rrset.List()` implementation to perform a paged walk

Some federated service e2e tests and a few ingress tests would become flaky after a few hundred runs. @csbell spent quite a lot of time debugging this and found out that this flakiness was due to a bug in the federated service controller deletion logic. Deletion of a federated service object triggers a logic in the controller to update the DNS records corresponding to that object. This DNS record update logic would return an error in failed runs which would in-turn cause the controller to reschedule the operation. This led to an infinite retry-failure cycle that never gave the API server a chance to garbage collect the deleted service object.

A couple of days ago we started seeing a correlation between the number of resource records in a DNS managed zone and these test failures. If you look at the test runs before and after run 2900 in the test grid - https://k8s-testgrid.appspot.com/cluster-federation#gce, you will notice that the grid became super green at 2900. That's when I deleted all the dangling DNS records from the past runs.

After some investigation yesterday, we found that `ResourceRecordSet.Get()` interface and its implementation, and `ResourceRecordSet.List()` implementation at least for Google Cloud DNS were incorrect.

This PR makes minimal set of changes (read: least invasive) in Google Cloud DNS provider implementation to fix these problems:

1. Modifies DNS provider Rrset.Get(name) interface to return multiple records and updates federated service controller.

    There can be multiple DNS resource records for a given name. They can vary by type, ttl, rrdata and a number of various other parameters. It is incorrect to return a single resource record for a given name.

    This change updates the Get interface to return multiple records for a given name and uses this list in the federated service controller to perform DNS operations.

2. Update Google Cloud DNS List implementation to perform a paged walk of lists to aggregate all the DNS records.

    The current `List()` implementation just lists the DNS resorce records in a given managed zone once and retruns the list. It neither performs a paged walk nor does it consider the `page_token` in the returned response.

    This change walks all the pages and aggregates the records in the pages and returns the aggregated list. This is potentially dangerous as it can blow up memory if there are a huge number of records in the given managed zone. But this is the best we can do without changing the provider interface too much. 

    Next step is to define a new paged list interface and implement it.

**Release note**:
```release-note
NONE
```

/assign @csbell 

cc @justinsb @shashidharatd @quinton-hoole @kubernetes/sig-federation-pr-reviews

federation/pkg/dnsprovider/dns.go

@@ -52,8 +52,12 @@ type Zone interface {
 type ResourceRecordSets interface {
 	// List returns the ResourceRecordSets of the Zone, or an error if the list operation failed.
 	List() ([]ResourceRecordSet, error)
-	// Get returns the ResourceRecordSet with the name in the Zone. if the named resource record set does not exist, but no error occurred, the returned set, and error, are both nil.
-	Get(name string) (ResourceRecordSet, error)
+	// Get returns the ResourceRecordSet list with the name in the Zone.
+	// This is a list because there might be multiple records of different
+	// types for a given name. If the named resource record sets do not
+	// exist, but no error occurred, the returned record set will be empty
+	// and error will be nil.
+	Get(name string) ([]ResourceRecordSet, error)
 	// New allocates a new ResourceRecordSet, which can then be passed to ResourceRecordChangeset Add() or Remove()
 	// Arguments are as per the ResourceRecordSet interface below.
 	New(name string, rrdatas []string, ttl int64, rrstype rrstype.RrsType) ResourceRecordSet

federation/pkg/dnsprovider/providers/aws/route53/rrsets.go

@@ -48,19 +48,30 @@ func (rrsets ResourceRecordSets) List() ([]dnsprovider.ResourceRecordSet, error)
 	return list, nil
 }
 
-func (rrsets ResourceRecordSets) Get(name string) (dnsprovider.ResourceRecordSet, error) {
-	var newRrset dnsprovider.ResourceRecordSet
-	rrsetList, err := rrsets.List()
+func (rrsets ResourceRecordSets) Get(name string) ([]dnsprovider.ResourceRecordSet, error) {
+	// This list implementation is very similar to the one implemented in
+	// the List() method above, but it restricts the retrieved list to
+	// the records whose name match the given `name`.
+	input := route53.ListResourceRecordSetsInput{
+		HostedZoneId:    rrsets.zone.impl.Id,
+		StartRecordName: aws.String(name),
+	}
+
+	var list []dnsprovider.ResourceRecordSet
+	err := rrsets.zone.zones.interface_.service.ListResourceRecordSetsPages(&input, func(page *route53.ListResourceRecordSetsOutput, lastPage bool) bool {
+		for _, rrset := range page.ResourceRecordSets {
+			if aws.StringValue(rrset.Name) != name {
+				return false
+			}
+			list = append(list, &ResourceRecordSet{rrset, &rrsets})
+		}
+		return true
+	})
 	if err != nil {
 		return nil, err
 	}
-	for _, rrset := range rrsetList {
-		if rrset.Name() == name {
-			newRrset = rrset
-			break
-		}
-	}
-	return newRrset, nil
+
+	return list, nil
 }
 
 func (r ResourceRecordSets) StartChangeset() dnsprovider.ResourceRecordChangeset {

federation/pkg/dnsprovider/providers/coredns/coredns_test.go

@@ -124,16 +124,16 @@ func listRrsOrFail(t *testing.T, rrsets dnsprovider.ResourceRecordSets) []dnspro
 	return rrset
 }
 
-func getRrOrFail(t *testing.T, rrsets dnsprovider.ResourceRecordSets, name string) dnsprovider.ResourceRecordSet {
-	rrset, err := rrsets.Get(name)
+func getRrOrFail(t *testing.T, rrsets dnsprovider.ResourceRecordSets, name string) []dnsprovider.ResourceRecordSet {
+	rrsetList, err := rrsets.Get(name)
 	if err != nil {
 		t.Fatalf("Failed to get recordset: %v", err)
-	} else if rrset == nil {
+	} else if len(rrsetList) == 0 {
 		t.Logf("Did not Get recordset: %v", name)
 	} else {
-		t.Logf("Got recordset: %v", rrset.Name())
+		t.Logf("Got recordsets: %v", rrsetList)
 	}
-	return rrset
+	return rrsetList
 }
 
 func getExampleRrs(zone dnsprovider.Zone) dnsprovider.ResourceRecordSet {

federation/pkg/dnsprovider/providers/coredns/rrsets.go

@@ -40,7 +40,7 @@ func (rrsets ResourceRecordSets) List() ([]dnsprovider.ResourceRecordSet, error)
 	return list, fmt.Errorf("OperationNotSupported")
 }
 
-func (rrsets ResourceRecordSets) Get(name string) (dnsprovider.ResourceRecordSet, error) {
+func (rrsets ResourceRecordSets) Get(name string) ([]dnsprovider.ResourceRecordSet, error) {
 	getOpts := &etcdc.GetOptions{
 		Recursive: true,
 	}
@@ -58,17 +58,16 @@ func (rrsets ResourceRecordSets) Get(name string) (dnsprovider.ResourceRecordSet
 		return nil, nil
 	}
 
-	rrset := ResourceRecordSet{name: name, rrdatas: []string{}, rrsets: &rrsets}
-	found := false
+	var list []dnsprovider.ResourceRecordSet
+
 	for _, node := range response.Node.Nodes {
-		found = true
 		service := dnsmsg.Service{}
 		err = json.Unmarshal([]byte(node.Value), &service)
 		if err != nil {
 			return nil, fmt.Errorf("Failed to unmarshall json data, err: %v", err)
 		}
 
-		// assuming all rrdatas in a rrset will have same type
+		rrset := ResourceRecordSet{name: name, rrdatas: []string{}, rrsets: &rrsets}
 		ip := net.ParseIP(service.Host)
 		switch {
 		case ip == nil:
@@ -78,13 +77,10 @@ func (rrsets ResourceRecordSets) Get(name string) (dnsprovider.ResourceRecordSet
 		}
 		rrset.rrdatas = append(rrset.rrdatas, service.Host)
 		rrset.ttl = int64(service.TTL)
+		list = append(list, rrset)
 	}
 
-	if !found {
-		return nil, nil
-	}
-
-	return rrset, nil
+	return list, nil
 }
 
 func (rrsets ResourceRecordSets) StartChangeset() dnsprovider.ResourceRecordChangeset {

federation/pkg/dnsprovider/providers/google/clouddns/internal/interfaces/interfaces.go

@@ -17,6 +17,8 @@ limitations under the License.
 package interfaces
 
 import (
+	"context"
+
 	"google.golang.org/api/googleapi"
 	"k8s.io/kubernetes/federation/pkg/dnsprovider/rrstype"
 )
@@ -172,8 +174,8 @@ type (
 
 	ResourceRecordSetsListCall interface {
 		// Context(ctx context.Context) *ResourceRecordSetsListCall  // TODO: Add as needed
-		// Do(opts ...googleapi.CallOption) (*ResourceRecordSetsListResponse, error)  // TODO: Add as needed
 		Do(opts ...googleapi.CallOption) (ResourceRecordSetsListResponse, error)
+		Pages(ctx context.Context, f func(ResourceRecordSetsListResponse) error) error
 		// Fields(s ...googleapi.Field) *ResourceRecordSetsListCall  // TODO: Add as needed
 		// IfNoneMatch(entityTag string) *ResourceRecordSetsListCall  // TODO: Add as needed
 		// MaxResults(maxResults int64) *ResourceRecordSetsListCall  // TODO: Add as needed
@@ -191,8 +193,10 @@ type (
 	}
 
 	ResourceRecordSetsService interface {
-		// NewResourceRecordSetsService(s *Service) *ResourceRecordSetsService // TODO: add to service as needed
 		List(project string, managedZone string) ResourceRecordSetsListCall
+		// Get returns a list of resources records with the matching name
+		Get(project, managedZone, name string) ResourceRecordSetsListCall
+		// NewResourceRecordSetsService(s *Service) *ResourceRecordSetsService // TODO: add to service as needed
 		NewResourceRecordSet(name string, rrdatas []string, ttl int64, type_ rrstype.RrsType) ResourceRecordSet
 	}
 

federation/pkg/dnsprovider/providers/google/clouddns/internal/rrsets_list_call.go

@@ -17,6 +17,8 @@ limitations under the License.
 package internal
 
 import (
+	"context"
+
 	dns "google.golang.org/api/dns/v1"
 	"google.golang.org/api/googleapi"
 	"k8s.io/kubernetes/federation/pkg/dnsprovider/providers/google/clouddns/internal/interfaces"
@@ -34,6 +36,12 @@ func (call *ResourceRecordSetsListCall) Do(opts ...googleapi.CallOption) (interf
 	return &ResourceRecordSetsListResponse{response}, err
 }
 
+func (call *ResourceRecordSetsListCall) Pages(ctx context.Context, f func(interfaces.ResourceRecordSetsListResponse) error) error {
+	return call.impl.Pages(ctx, func(page *dns.ResourceRecordSetsListResponse) error {
+		return f(&ResourceRecordSetsListResponse{page})
+	})
+}
+
 func (call *ResourceRecordSetsListCall) Name(name string) interfaces.ResourceRecordSetsListCall {
 	call.impl.Name(name)
 	return call

federation/pkg/dnsprovider/providers/google/clouddns/internal/rrsets_service.go

@@ -33,6 +33,10 @@ func (service ResourceRecordSetsService) List(project string, managedZone string
 	return &ResourceRecordSetsListCall{service.impl.List(project, managedZone)}
 }
 
+func (service ResourceRecordSetsService) Get(project, managedZone, name string) interfaces.ResourceRecordSetsListCall {
+	return &ResourceRecordSetsListCall{service.impl.List(project, managedZone).Name(name)}
+}
+
 func (service ResourceRecordSetsService) NewResourceRecordSet(name string, rrdatas []string, ttl int64, type_ rrstype.RrsType) interfaces.ResourceRecordSet {
 	rrset := dns.ResourceRecordSet{Name: name, Rrdatas: rrdatas, Ttl: ttl, Type: string(type_)}
 	return &ResourceRecordSet{&rrset}

federation/pkg/dnsprovider/providers/google/clouddns/internal/stubs/rrsets_list_call.go

@@ -17,6 +17,8 @@ limitations under the License.
 package stubs
 
 import (
+	"context"
+
 	"google.golang.org/api/googleapi"
 	"k8s.io/kubernetes/federation/pkg/dnsprovider/providers/google/clouddns/internal/interfaces"
 )
@@ -35,6 +37,10 @@ func (call *ResourceRecordSetsListCall) Do(opts ...googleapi.CallOption) (interf
 	return call.Response_, call.Err_
 }
 
+func (call *ResourceRecordSetsListCall) Pages(ctx context.Context, f func(interfaces.ResourceRecordSetsListResponse) error) error {
+	return f(call.Response_)
+}
+
 func (call *ResourceRecordSetsListCall) Name(name string) interfaces.ResourceRecordSetsListCall {
 	call.Name_ = name
 	return call

federation/pkg/dnsprovider/providers/google/clouddns/internal/stubs/rrsets_service.go

@@ -31,21 +31,27 @@ type ResourceRecordSetsService struct {
 	ListCall interfaces.ResourceRecordSetsListCall // Use to override response if required for testing
 }
 
+func (s ResourceRecordSetsService) managedZone(project, managedZone string) (*ManagedZone, error) {
+	p := s.Service.ManagedZones_.Impl[project]
+	if p == nil {
+		return nil, fmt.Errorf("Project not found: %s", project)
+	}
+	z := s.Service.ManagedZones_.Impl[project][managedZone]
+	if z == nil {
+		return nil, fmt.Errorf("Zone %s not found in project %s", managedZone, project)
+	}
+	return z.(*ManagedZone), nil
+}
+
 func (s ResourceRecordSetsService) List(project string, managedZone string) interfaces.ResourceRecordSetsListCall {
 	if s.ListCall != nil {
 		return s.ListCall
 	}
-	p := s.Service.ManagedZones_.Impl[project]
-	if p == nil {
-		return &ResourceRecordSetsListCall{Err_: fmt.Errorf("Project not found: %s", project)}
+	zone, err := s.managedZone(project, managedZone)
+	if err != nil {
+		return &ResourceRecordSetsListCall{Err_: err}
 	}
-	z := s.Service.ManagedZones_.Impl[project][managedZone]
-	if z == nil {
-		return &ResourceRecordSetsListCall{
-			Err_: fmt.Errorf("Zone %s not found in project %s", managedZone, project),
-		}
-	}
-	zone := s.Service.ManagedZones_.Impl[project][managedZone].(*ManagedZone)
+
 	response := &ResourceRecordSetsListResponse{}
 	for _, set := range zone.Rrsets {
 		response.impl = append(response.impl, set)
@@ -53,6 +59,24 @@ func (s ResourceRecordSetsService) List(project string, managedZone string) inte
 	return &ResourceRecordSetsListCall{Response_: response}
 }
 
+func (s ResourceRecordSetsService) Get(project, managedZone, name string) interfaces.ResourceRecordSetsListCall {
+	if s.ListCall != nil {
+		return s.ListCall
+	}
+	zone, err := s.managedZone(project, managedZone)
+	if err != nil {
+		return &ResourceRecordSetsListCall{Err_: err}
+	}
+
+	response := &ResourceRecordSetsListResponse{}
+	for _, set := range zone.Rrsets {
+		if set.Name_ == name {
+			response.impl = append(response.impl, set)
+		}
+	}
+	return &ResourceRecordSetsListCall{Response_: response}
+}
+
 func (service ResourceRecordSetsService) NewResourceRecordSet(name string, rrdatas []string, ttl int64, type_ rrstype.RrsType) interfaces.ResourceRecordSet {
 	rrset := ResourceRecordSet{Name_: name, Rrdatas_: rrdatas, Ttl_: ttl, Type_: string(type_)}
 	return rrset

federation/pkg/dnsprovider/providers/google/clouddns/rrsets.go

@@ -17,6 +17,8 @@ limitations under the License.
 package clouddns
 
 import (
+	"context"
+
 	"k8s.io/kubernetes/federation/pkg/dnsprovider"
 	"k8s.io/kubernetes/federation/pkg/dnsprovider/providers/google/clouddns/internal/interfaces"
 	"k8s.io/kubernetes/federation/pkg/dnsprovider/rrstype"
@@ -30,31 +32,46 @@ type ResourceRecordSets struct {
 	impl interfaces.ResourceRecordSetsService
 }
 
+// List returns a list of resource records in the given project and
+// managed zone.
+// !!CAUTION!! Your memory might explode if you have a huge number of
+// records in your managed zone.
 func (rrsets ResourceRecordSets) List() ([]dnsprovider.ResourceRecordSet, error) {
-	response, err := rrsets.impl.List(rrsets.project(), rrsets.zone.impl.Name()).Do()
+	var list []dnsprovider.ResourceRecordSet
+
+	ctx := context.Background()
+
+	call := rrsets.impl.List(rrsets.project(), rrsets.zone.impl.Name())
+	err := call.Pages(ctx, func(page interfaces.ResourceRecordSetsListResponse) error {
+		for _, rrset := range page.Rrsets() {
+			list = append(list, ResourceRecordSet{rrset, &rrsets})
+		}
+		return nil
+	})
 	if err != nil {
 		return nil, err
 	}
-	list := make([]dnsprovider.ResourceRecordSet, len(response.Rrsets()))
-	for i, rrset := range response.Rrsets() {
-		list[i] = ResourceRecordSet{rrset, &rrsets}
-	}
+
 	return list, nil
 }
 
-func (rrsets ResourceRecordSets) Get(name string) (dnsprovider.ResourceRecordSet, error) {
-	var newRrset dnsprovider.ResourceRecordSet
-	rrsetList, err := rrsets.List()
+func (rrsets ResourceRecordSets) Get(name string) ([]dnsprovider.ResourceRecordSet, error) {
+	var list []dnsprovider.ResourceRecordSet
+
+	ctx := context.Background()
+
+	call := rrsets.impl.Get(rrsets.project(), rrsets.zone.impl.Name(), name)
+	err := call.Pages(ctx, func(page interfaces.ResourceRecordSetsListResponse) error {
+		for _, rrset := range page.Rrsets() {
+			list = append(list, ResourceRecordSet{rrset, &rrsets})
+		}
+		return nil
+	})
 	if err != nil {
 		return nil, err
 	}
-	for _, rrset := range rrsetList {
-		if rrset.Name() == name {
-			newRrset = rrset
-			break
-		}
-	}
-	return newRrset, nil
+
+	return list, nil
 }
 
 func (r ResourceRecordSets) StartChangeset() dnsprovider.ResourceRecordChangeset {

federation/pkg/dnsprovider/tests/commontests.go

@@ -108,16 +108,16 @@ func rrs(t *testing.T, zone dnsprovider.Zone) (r dnsprovider.ResourceRecordSets)
 	return rrsets
 }
 
-func getRrOrFail(t *testing.T, rrsets dnsprovider.ResourceRecordSets, name string) dnsprovider.ResourceRecordSet {
-	rrset, err := rrsets.Get(name)
+func getRrOrFail(t *testing.T, rrsets dnsprovider.ResourceRecordSets, name string) []dnsprovider.ResourceRecordSet {
+	rrsetList, err := rrsets.Get(name)
 	if err != nil {
 		t.Fatalf("Failed to get recordset: %v", err)
-	} else if rrset == nil {
+	} else if len(rrsetList) == 0 {
 		t.Logf("Did not Get recordset: %v", name)
 	} else {
-		t.Logf("Got recordset: %v", rrset.Name())
+		t.Logf("Got recordset: %v", rrsetList[0].Name())
 	}
-	return rrset
+	return rrsetList
 }
 
 // assertHasRecord tests that rrsets has a record equivalent to rrset
@@ -127,7 +127,13 @@ func assertHasRecord(t *testing.T, rrsets dnsprovider.ResourceRecordSets, rrset
 	rrs, err := rrsets.List()
 	if err != nil {
 		if err.Error() == "OperationNotSupported" {
-			found = getRrOrFail(t, rrsets, rrset.Name())
+			foundList := getRrOrFail(t, rrsets, rrset.Name())
+			for i, elem := range foundList {
+				if elem.Name() == rrset.Name() && elem.Type() == rrset.Type() {
+					found = foundList[i]
+					break
+				}
+			}
 		} else {
 			t.Fatalf("Failed to list recordsets: %v", err)
 		}

federation/pkg/federation-controller/service/dns.go

@@ -154,11 +154,22 @@ func getDnsZone(dnsZoneName string, dnsZoneID string, dnsZonesInterface dnsprovi
 	}
 }
 
-//   Note that if the named resource record set does not exist, but no error occurred, the returned set, and error, are both nil
-func getRrset(dnsName string, rrsetsInterface dnsprovider.ResourceRecordSets) (dnsprovider.ResourceRecordSet, error) {
+// NOTE: that if the named resource record set does not exist, but no
+// error occurred, the returned list will be empty, and the error will
+// be nil
+func getRrset(dnsName string, rrsetsInterface dnsprovider.ResourceRecordSets) ([]dnsprovider.ResourceRecordSet, error) {
 	return rrsetsInterface.Get(dnsName)
 }
 
+func findRrset(list []dnsprovider.ResourceRecordSet, rrset dnsprovider.ResourceRecordSet) dnsprovider.ResourceRecordSet {
+	for i, elem := range list {
+		if dnsprovider.ResourceRecordSetsEquivalent(rrset, elem) {
+			return list[i]
+		}
+	}
+	return nil
+}
+
 /* getResolvedEndpoints performs DNS resolution on the provided slice of endpoints (which might be DNS names or IPv4 addresses)
    and returns a list of IPv4 addresses.  If any of the endpoints are neither valid IPv4 addresses nor resolvable DNS names,
    non-nil error is also returned (possibly along with a partially complete list of resolved endpoints.
@@ -190,11 +201,11 @@ func (s *ServiceController) ensureDnsRrsets(dnsZone dnsprovider.Zone, dnsName st
 	if !supported {
 		return fmt.Errorf("Failed to ensure DNS records for %s. DNS provider does not support the ResourceRecordSets interface.", dnsName)
 	}
-	rrset, err := getRrset(dnsName, rrsets) // TODO: rrsets.Get(dnsName)
+	rrsetList, err := getRrset(dnsName, rrsets) // TODO: rrsets.Get(dnsName)
 	if err != nil {
 		return err
 	}
-	if rrset == nil {
+	if len(rrsetList) == 0 {
 		glog.V(4).Infof("No recordsets found for DNS name %q.  Need to add either A records (if we have healthy endpoints), or a CNAME record to %q", dnsName, uplevelCname)
 		if len(endpoints) < 1 {
 			glog.V(4).Infof("There are no healthy endpoint addresses at level %q, so CNAME to %q, if provided", dnsName, uplevelCname)
@@ -228,57 +239,66 @@ func (s *ServiceController) ensureDnsRrsets(dnsZone dnsprovider.Zone, dnsName st
 			glog.V(4).Infof("Successfully added recordset %v", newRrset)
 		}
 	} else {
-		// the rrset already exists, so make it right.
-		glog.V(4).Infof("Recordset %v already exists.  Ensuring that it is correct.", rrset)
+		// the rrsets already exists, so make it right.
+		glog.V(4).Infof("Recordset %v already exists. Ensuring that it is correct.", rrsetList)
 		if len(endpoints) < 1 {
 			// Need an appropriate CNAME record.  Check that we have it.
 			newRrset := rrsets.New(dnsName, []string{uplevelCname}, minDnsTtl, rrstype.CNAME)
-			glog.V(4).Infof("No healthy endpoints for %s.  Have recordset %v. Need recordset %v", dnsName, rrset, newRrset)
-			if dnsprovider.ResourceRecordSetsEquivalent(rrset, newRrset) {
+			glog.V(4).Infof("No healthy endpoints for %s. Have recordsets %v. Need recordset %v", dnsName, rrsetList, newRrset)
+			found := findRrset(rrsetList, newRrset)
+			if found != nil {
 				// The existing rrset is equivalent to the required one - our work is done here
-				glog.V(4).Infof("Existing recordset %v is equivalent to needed recordset %v, our work is done here.", rrset, newRrset)
+				glog.V(4).Infof("Existing recordset %v is equivalent to needed recordset %v, our work is done here.", rrsetList, newRrset)
 				return nil
 			} else {
 				// Need to replace the existing one with a better one (or just remove it if we have no healthy endpoints).
-				glog.V(4).Infof("Existing recordset %v not equivalent to needed recordset %v removing existing and adding needed.", rrset, newRrset)
+				glog.V(4).Infof("Existing recordset %v not equivalent to needed recordset %v removing existing and adding needed.", rrsetList, newRrset)
 				changeSet := rrsets.StartChangeset()
-				changeSet.Remove(rrset)
+				for i := range rrsetList {
+					changeSet = changeSet.Remove(rrsetList[i])
+				}
 				if uplevelCname != "" {
-					changeSet.Add(newRrset)
+					changeSet = changeSet.Add(newRrset)
 					if err := changeSet.Apply(); err != nil {
 						return err
 					}
-					glog.V(4).Infof("Successfully replaced needed recordset %v -> %v", rrset, newRrset)
+					glog.V(4).Infof("Successfully replaced needed recordset %v -> %v", found, newRrset)
 				} else {
 					if err := changeSet.Apply(); err != nil {
 						return err
 					}
-					glog.V(4).Infof("Successfully removed existing recordset %v", rrset)
+					glog.V(4).Infof("Successfully removed existing recordset %v", found)
 					glog.V(4).Infof("Uplevel CNAME is empty string. Not adding recordset %v", newRrset)
 				}
 			}
 		} else {
 			// We have an rrset in DNS, possibly with some missing addresses and some unwanted addresses.
 			// And we have healthy endpoints.  Just replace what's there with the healthy endpoints, if it's not already correct.
-			glog.V(4).Infof("%s: Healthy endpoints %v exist.  Recordset %v exists.  Reconciling.", dnsName, endpoints, rrset)
+			glog.V(4).Infof("%s: Healthy endpoints %v exist. Recordset %v exists.  Reconciling.", dnsName, endpoints, rrsetList)
 			resolvedEndpoints, err := getResolvedEndpoints(endpoints)
 			if err != nil { // Some invalid addresses or otherwise unresolvable DNS names.
 				return err // TODO: We could potentially add the ones we did get back, even if some of them failed to resolve.
 			}
 			newRrset := rrsets.New(dnsName, resolvedEndpoints, minDnsTtl, rrstype.A)
-			glog.V(4).Infof("Have recordset %v. Need recordset %v", rrset, newRrset)
-			if dnsprovider.ResourceRecordSetsEquivalent(rrset, newRrset) {
-				glog.V(4).Infof("Existing recordset %v is equivalent to needed recordset %v, our work is done here.", rrset, newRrset)
+			glog.V(4).Infof("Have recordset %v. Need recordset %v", rrsetList, newRrset)
+			found := findRrset(rrsetList, newRrset)
+			if found != nil {
+				glog.V(4).Infof("Existing recordset %v is equivalent to needed recordset %v, our work is done here.", found, newRrset)
 				// TODO: We could be more thorough about checking for equivalence to avoid unnecessary updates, but in the
 				//       worst case we'll just replace what's there with an equivalent, if not exactly identical record set.
 				return nil
 			} else {
 				// Need to replace the existing one with a better one
-				glog.V(4).Infof("Existing recordset %v is not equivalent to needed recordset %v, removing existing and adding needed.", rrset, newRrset)
-				if err = rrsets.StartChangeset().Remove(rrset).Add(newRrset).Apply(); err != nil {
+				glog.V(4).Infof("Existing recordset %v is not equivalent to needed recordset %v, removing existing and adding needed.", found, newRrset)
+				changeSet := rrsets.StartChangeset()
+				for i := range rrsetList {
+					changeSet = changeSet.Remove(rrsetList[i])
+				}
+				changeSet = changeSet.Add(newRrset)
+				if err = changeSet.Apply(); err != nil {
 					return err
 				}
-				glog.V(4).Infof("Successfully replaced recordset %v -> %v", rrset, newRrset)
+				glog.V(4).Infof("Successfully replaced recordset %v -> %v", found, newRrset)
 			}
 		}
 	}