Reduce AWS Route53 API calls

Currently, planning instructs to create all records even
those which does not match any zone.
Later, those records will be checked towards the existing
records and filtered whether they match or not a hosted zone.

This causes a problem, at least in the specific case of the Route53
implementation as it always calls the ApplyChanges method, which in its
turn always retrieves all records in all zones.

This causes high pressure on Route53 APIs, for non-necessary actions.

By being able to filter all unmanaged records from the plan, we can
prevent from calling ApplyChanges when nothing has to be done and hence
prevent an unnecessary listing of records.

By doing so, the rate of API calls to AWS Route53 is expected to be
reduced by 2

controller/controller.go

@@ -72,6 +72,14 @@ var (
 			Help:      "Timestamp of last successful sync with the DNS provider",
 		},
 	)
+	controllerNoChangesTotal = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: "external_dns",
+			Subsystem: "controller",
+			Name:      "no_op_runs_total",
+			Help:      "Number of reconcile loops ending up with no changes on the DNS provider side.",
+		},
+	)
 	deprecatedRegistryErrors = prometheus.NewCounter(
 		prometheus.CounterOpts{
 			Subsystem: "registry",
@@ -96,6 +104,7 @@ func init() {
 	prometheus.MustRegister(lastSyncTimestamp)
 	prometheus.MustRegister(deprecatedRegistryErrors)
 	prometheus.MustRegister(deprecatedSourceErrors)
+	prometheus.MustRegister(controllerNoChangesTotal)
 }
 
 // Controller is responsible for orchestrating the different components.
@@ -112,7 +121,7 @@ type Controller struct {
 	// The interval between individual synchronizations
 	Interval time.Duration
 	// The DomainFilter defines which DNS records to keep or exclude
-	DomainFilter endpoint.DomainFilter
+	DomainFilter endpoint.DomainFilterInterface
 	// The nextRunAt used for throttling and batching reconciliation
 	nextRunAt time.Time
 	// The nextRunAtMux is for atomic updating of nextRunAt
@@ -144,23 +153,29 @@ func (c *Controller) RunOnce(ctx context.Context) error {
 	sourceEndpointsTotal.Set(float64(len(endpoints)))
 
 	endpoints = c.Registry.AdjustEndpoints(endpoints)
+	//filter :=
 
 	plan := &plan.Plan{
 		Policies:           []plan.Policy{c.Policy},
 		Current:            records,
 		Desired:            endpoints,
-		DomainFilter:       c.DomainFilter,
+		DomainFilter:       endpoint.MatchAllDomainFilters{c.DomainFilter, c.Registry.GetDomainFilter()},
 		PropertyComparator: c.Registry.PropertyValuesEqual,
 		ManagedRecords:     []string{endpoint.RecordTypeA, endpoint.RecordTypeCNAME},
 	}
 
 	plan = plan.Calculate()
 
-	err = c.Registry.ApplyChanges(ctx, plan.Changes)
-	if err != nil {
-		registryErrorsTotal.Inc()
-		deprecatedRegistryErrors.Inc()
-		return err
+	if plan.Changes.HasChanges() {
+		err = c.Registry.ApplyChanges(ctx, plan.Changes)
+		if err != nil {
+			registryErrorsTotal.Inc()
+			deprecatedRegistryErrors.Inc()
+			return err
+		}
+	} else {
+		controllerNoChangesTotal.Inc()
+		log.Info("All records are already up to date")
 	}
 
 	lastSyncTimestamp.SetToCurrentTime()

controller/controller_test.go

@@ -40,6 +40,30 @@ type mockProvider struct {
 	ExpectChanges *plan.Changes
 }
 
+type filteredMockProvider struct {
+	provider.BaseProvider
+	domainFilter      endpoint.DomainFilterInterface
+	RecordsStore      []*endpoint.Endpoint
+	RecordsCallCount  int
+	ApplyChangesCalls []*plan.Changes
+}
+
+func (p *filteredMockProvider) GetDomainFilter() endpoint.DomainFilterInterface {
+	return p.domainFilter
+}
+
+// Records returns the desired mock endpoints.
+func (p *filteredMockProvider) Records(ctx context.Context) ([]*endpoint.Endpoint, error) {
+	p.RecordsCallCount++
+	return p.RecordsStore, nil
+}
+
+// ApplyChanges stores all calls for later check
+func (p *filteredMockProvider) ApplyChanges(ctx context.Context, changes *plan.Changes) error {
+	p.ApplyChangesCalls = append(p.ApplyChangesCalls, changes)
+	return nil
+}
+
 // Records returns the desired mock endpoints.
 func (p *mockProvider) Records(ctx context.Context) ([]*endpoint.Endpoint, error) {
 	return p.RecordsStore, nil
@@ -192,3 +216,155 @@ func TestShouldRunOnce(t *testing.T) {
 	// But not two times
 	assert.False(t, ctrl.ShouldRunOnce(now))
 }
+
+func testControllerFiltersDomains(t *testing.T, configuredEndpoints []*endpoint.Endpoint, domainFilter endpoint.DomainFilterInterface, providerEndpoints []*endpoint.Endpoint, expectedChanges []*plan.Changes) {
+	t.Helper()
+	source := new(testutils.MockSource)
+	source.On("Endpoints").Return(configuredEndpoints, nil)
+
+	// Fake some existing records in our DNS provider and validate some desired changes.
+	provider := &filteredMockProvider{
+		RecordsStore: providerEndpoints,
+	}
+	r, err := registry.NewNoopRegistry(provider)
+
+	require.NoError(t, err)
+
+	ctrl := &Controller{
+		Source:       source,
+		Registry:     r,
+		Policy:       &plan.SyncPolicy{},
+		DomainFilter: domainFilter,
+	}
+
+	assert.NoError(t, ctrl.RunOnce(context.Background()))
+	assert.Equal(t, 1, provider.RecordsCallCount)
+	require.Len(t, provider.ApplyChangesCalls, len(expectedChanges))
+	for i, change := range expectedChanges {
+		assert.Equal(t, *change, *provider.ApplyChangesCalls[i])
+	}
+}
+
+func TestControllerSkipsEmptyChanges(t *testing.T) {
+	testControllerFiltersDomains(
+		t,
+		[]*endpoint.Endpoint{
+			{
+				DNSName:    "create-record.other.tld",
+				RecordType: endpoint.RecordTypeA,
+				Targets:    endpoint.Targets{"1.2.3.4"},
+			},
+			{
+				DNSName:    "some-record.used.tld",
+				RecordType: endpoint.RecordTypeA,
+				Targets:    endpoint.Targets{"8.8.8.8"},
+			},
+		},
+		endpoint.NewDomainFilter([]string{"used.tld"}),
+		[]*endpoint.Endpoint{
+			{
+				DNSName:    "some-record.used.tld",
+				RecordType: endpoint.RecordTypeA,
+				Targets:    endpoint.Targets{"8.8.8.8"},
+			},
+		},
+		[]*plan.Changes{},
+	)
+}
+
+func TestWhenNoFilterControllerConsidersAllComain(t *testing.T) {
+	testControllerFiltersDomains(
+		t,
+		[]*endpoint.Endpoint{
+			{
+				DNSName:    "create-record.other.tld",
+				RecordType: endpoint.RecordTypeA,
+				Targets:    endpoint.Targets{"1.2.3.4"},
+			},
+			{
+				DNSName:    "some-record.used.tld",
+				RecordType: endpoint.RecordTypeA,
+				Targets:    endpoint.Targets{"8.8.8.8"},
+			},
+		},
+		nil,
+		[]*endpoint.Endpoint{
+			{
+				DNSName:    "some-record.used.tld",
+				RecordType: endpoint.RecordTypeA,
+				Targets:    endpoint.Targets{"8.8.8.8"},
+			},
+		},
+		[]*plan.Changes{
+			{
+				Create: []*endpoint.Endpoint{
+					{
+						DNSName:    "create-record.other.tld",
+						RecordType: endpoint.RecordTypeA,
+						Targets:    endpoint.Targets{"1.2.3.4"},
+					},
+				},
+			},
+		},
+	)
+}
+
+func TestWhenMultipleControllerConsidersAllFilteredComain(t *testing.T) {
+	testControllerFiltersDomains(
+		t,
+		[]*endpoint.Endpoint{
+			{
+				DNSName:    "create-record.other.tld",
+				RecordType: endpoint.RecordTypeA,
+				Targets:    endpoint.Targets{"1.2.3.4"},
+			},
+			{
+				DNSName:    "some-record.used.tld",
+				RecordType: endpoint.RecordTypeA,
+				Targets:    endpoint.Targets{"1.1.1.1"},
+			},
+			{
+				DNSName:    "create-record.unused.tld",
+				RecordType: endpoint.RecordTypeA,
+				Targets:    endpoint.Targets{"1.2.3.4"},
+			},
+		},
+		endpoint.NewDomainFilter([]string{"used.tld", "other.tld"}),
+		[]*endpoint.Endpoint{
+			{
+				DNSName:    "some-record.used.tld",
+				RecordType: endpoint.RecordTypeA,
+				Targets:    endpoint.Targets{"8.8.8.8"},
+			},
+		},
+		[]*plan.Changes{
+			{
+				Create: []*endpoint.Endpoint{
+					{
+						DNSName:    "create-record.other.tld",
+						RecordType: endpoint.RecordTypeA,
+						Targets:    endpoint.Targets{"1.2.3.4"},
+					},
+				},
+				UpdateOld: []*endpoint.Endpoint{
+					{
+						DNSName:    "some-record.used.tld",
+						RecordType: endpoint.RecordTypeA,
+						Targets:    endpoint.Targets{"8.8.8.8"},
+						Labels:     endpoint.Labels{},
+					},
+				},
+				UpdateNew: []*endpoint.Endpoint{
+					{
+						DNSName:    "some-record.used.tld",
+						RecordType: endpoint.RecordTypeA,
+						Targets:    endpoint.Targets{"1.1.1.1"},
+						Labels: endpoint.Labels{
+							"owner": "",
+						},
+					},
+				},
+			},
+		},
+	)
+}

endpoint/domain_filter.go

@@ -21,6 +21,44 @@ import (
 	"strings"
 )
 
+// DomainFilterInterface defines the interface to select matching domains for a specific provider or runtime
+type DomainFilterInterface interface {
+	Match(domain string) bool
+	IsConfigured() bool
+}
+
+type MatchAllDomainFilters []DomainFilterInterface
+
+func (f MatchAllDomainFilters) Match(domain string) bool {
+	if !f.IsConfigured() {
+		return true
+	}
+	for _, filter := range f {
+		if filter == nil {
+			continue
+		}
+		if filter.IsConfigured() && !filter.Match(domain) {
+			return false
+		}
+	}
+	return true
+}
+
+func (f MatchAllDomainFilters) IsConfigured() bool {
+	if f == nil {
+		return false
+	}
+	for _, filter := range f {
+		if filter == nil {
+			continue
+		}
+		if filter.IsConfigured() {
+			return true
+		}
+	}
+	return len(f) > 0
+}
+
 // DomainFilter holds a lists of valid domain names
 type DomainFilter struct {
 	// Filters define what domains to match

plan/plan.go

@@ -21,6 +21,9 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/google/go-cmp/cmp"
+	log "github.com/sirupsen/logrus"
+
 	"sigs.k8s.io/external-dns/endpoint"
 )
 
@@ -40,7 +43,7 @@ type Plan struct {
 	// Populated after calling Calculate()
 	Changes *Changes
 	// DomainFilter matches DNS names
-	DomainFilter endpoint.DomainFilter
+	DomainFilter endpoint.DomainFilterInterface
 	// Property comparator compares custom properties of providers
 	PropertyComparator PropertyComparator
 	// DNS record types that will be considered for management
@@ -115,12 +118,23 @@ func (t planTable) addCandidate(e *endpoint.Endpoint) {
 	t.rows[dnsName][e.SetIdentifier].candidates = append(t.rows[dnsName][e.SetIdentifier].candidates, e)
 }
 
+func (c *Changes) HasChanges() bool {
+	if len(c.Create) > 0 || len(c.Delete) > 0 {
+		return true
+	}
+	return !cmp.Equal(c.UpdateNew, c.UpdateOld)
+}
+
 // Calculate computes the actions needed to move current state towards desired
 // state. It then passes those changes to the current policy for further
 // processing. It returns a copy of Plan with the changes populated.
 func (p *Plan) Calculate() *Plan {
 	t := newPlanTable()
 
+	if p.DomainFilter == nil {
+		p.DomainFilter = endpoint.MatchAllDomainFilters(nil)
+	}
+
 	for _, current := range filterRecordsForPlan(p.Current, p.DomainFilter, p.ManagedRecords) {
 		t.addCurrent(current)
 	}
@@ -227,12 +241,13 @@ func (p *Plan) shouldUpdateProviderSpecific(desired, current *endpoint.Endpoint)
 // Per RFC 1034, CNAME records conflict with all other records - it is the
 // only record with this property. The behavior of the planner may need to be
 // made more sophisticated to codify this.
-func filterRecordsForPlan(records []*endpoint.Endpoint, domainFilter endpoint.DomainFilter, managedRecords []string) []*endpoint.Endpoint {
+func filterRecordsForPlan(records []*endpoint.Endpoint, domainFilter endpoint.DomainFilterInterface, managedRecords []string) []*endpoint.Endpoint {
 	filtered := []*endpoint.Endpoint{}
 
 	for _, record := range records {
 		// Ignore records that do not match the domain filter provided
 		if !domainFilter.Match(record.DNSName) {
+			log.Debugf("ignoring record %s that does not match domain filter", record.DNSName)
 			continue
 		}
 		if isManagedRecord(record.RecordType, managedRecords) {

provider/aws/aws.go

@@ -465,6 +465,21 @@ func (p *AWSProvider) createUpdateChanges(newEndpoints, oldEndpoints []*endpoint
 	return combined
 }
 
+// GetDomainFilter generates a filter to exclude any domain that is not controlled by the provider
+func (p *AWSProvider) GetDomainFilter() endpoint.DomainFilterInterface {
+	zones, err := p.Zones(context.Background())
+	if err != nil {
+		log.Errorf("failed to list zones: %v", err)
+		return &endpoint.DomainFilter{}
+	}
+	zoneNames := []string(nil)
+	for _, z := range zones {
+		zoneNames = append(zoneNames, aws.StringValue(z.Name), "."+aws.StringValue(z.Name))
+	}
+	log.Infof("Applying provider record filter for domains: %v", zoneNames)
+	return endpoint.NewDomainFilter(zoneNames)
+}
+
 // ApplyChanges applies a given set of changes in a given zone.
 func (p *AWSProvider) ApplyChanges(ctx context.Context, changes *plan.Changes) error {
 	zones, err := p.Zones(ctx)

provider/aws/aws_test.go

@@ -313,6 +313,29 @@ func TestAWSZones(t *testing.T) {
 	}
 }
 
+func TestAWSRecordsFilter(t *testing.T) {
+	provider, _ := newAWSProvider(t, endpoint.DomainFilter{}, provider.ZoneIDFilter{}, provider.ZoneTypeFilter{}, false, false, nil)
+	domainFilter := provider.GetDomainFilter()
+	assert.NotNil(t, domainFilter)
+	require.IsType(t, endpoint.DomainFilter{}, domainFilter)
+	count := 0
+	filters := domainFilter.(endpoint.DomainFilter).Filters
+	for _, tld := range []string{
+		"zone-4.ext-dns-test-3.teapot.zalan.do",
+		".zone-4.ext-dns-test-3.teapot.zalan.do",
+		"zone-2.ext-dns-test-2.teapot.zalan.do",
+		".zone-2.ext-dns-test-2.teapot.zalan.do",
+		"zone-3.ext-dns-test-2.teapot.zalan.do",
+		".zone-3.ext-dns-test-2.teapot.zalan.do",
+		"zone-4.ext-dns-test-3.teapot.zalan.do",
+		".zone-4.ext-dns-test-3.teapot.zalan.do",
+	} {
+		assert.Contains(t, filters, tld)
+		count++
+	}
+	assert.Len(t, filters, count)
+}
+
 func TestAWSRecords(t *testing.T) {
 	provider, _ := newAWSProvider(t, endpoint.NewDomainFilter([]string{"ext-dns-test-2.teapot.zalan.do."}), provider.NewZoneIDFilter([]string{}), provider.NewZoneTypeFilter(""), false, false, []*endpoint.Endpoint{
 		endpoint.NewEndpointWithTTL("list-test.zone-1.ext-dns-test-2.teapot.zalan.do", endpoint.RecordTypeA, endpoint.TTL(recordTTL), "1.2.3.4"),

provider/provider.go

@@ -31,6 +31,7 @@ type Provider interface {
 	ApplyChanges(ctx context.Context, changes *plan.Changes) error
 	PropertyValuesEqual(name string, previous string, current string) bool
 	AdjustEndpoints(endpoints []*endpoint.Endpoint) []*endpoint.Endpoint
+	GetDomainFilter() endpoint.DomainFilterInterface
 }
 
 type BaseProvider struct {
@@ -44,6 +45,10 @@ func (b BaseProvider) PropertyValuesEqual(name, previous, current string) bool {
 	return previous == current
 }
 
+func (b BaseProvider) GetDomainFilter() endpoint.DomainFilterInterface {
+	return endpoint.DomainFilter{}
+}
+
 type contextKey struct {
 	name string
 }

registry/aws_sd_registry.go

@@ -42,6 +42,10 @@ func NewAWSSDRegistry(provider provider.Provider, ownerID string) (*AWSSDRegistr
 	}, nil
 }
 
+func (sdr *AWSSDRegistry) GetDomainFilter() endpoint.DomainFilterInterface {
+	return sdr.provider.GetDomainFilter()
+}
+
 // Records calls AWS SD API and expects AWS SD provider to provider Owner/Resource information as a serialized
 // value in the AWSSDDescriptionLabel value in the Labels map
 func (sdr *AWSSDRegistry) Records(ctx context.Context) ([]*endpoint.Endpoint, error) {

registry/noop.go

@@ -36,6 +36,10 @@ func NewNoopRegistry(provider provider.Provider) (*NoopRegistry, error) {
 	}, nil
 }
 
+func (im *NoopRegistry) GetDomainFilter() endpoint.DomainFilterInterface {
+	return im.provider.GetDomainFilter()
+}
+
 // Records returns the current records from the dns provider
 func (im *NoopRegistry) Records(ctx context.Context) ([]*endpoint.Endpoint, error) {
 	return im.provider.Records(ctx)

registry/registry.go

@@ -34,6 +34,7 @@ type Registry interface {
 	ApplyChanges(ctx context.Context, changes *plan.Changes) error
 	PropertyValuesEqual(attribute string, previous string, current string) bool
 	AdjustEndpoints(endpoints []*endpoint.Endpoint) []*endpoint.Endpoint
+	GetDomainFilter() endpoint.DomainFilterInterface
 }
 
 //TODO(ideahitme): consider moving this to Plan

registry/txt.go

@@ -68,6 +68,10 @@ func NewTXTRegistry(provider provider.Provider, txtPrefix, txtSuffix, ownerID st
 	}, nil
 }
 
+func (im *TXTRegistry) GetDomainFilter() endpoint.DomainFilterInterface {
+	return im.provider.GetDomainFilter()
+}
+
 // Records returns the current records from the registry excluding TXT Records
 // If TXT records was created previously to indicate ownership its corresponding value
 // will be added to the endpoints Labels map