Only store endpoints with their labels in the cache (#612)

* Set cacheinterval flag to 0 by default and if it is zero don't use cache

* fix: run gofmt -s in go1.11

delivery.yaml

@@ -9,10 +9,11 @@ build_steps:
     apt-get install -y docker-ce
 - desc: Build and push docker image
   cmd: |
-    image=registry-write.opensource.zalan.do/teapot/external-dns:$(git describe --always --dirty --tags)
-    docker build --squash --tag $image .
-    IS_PR_BUILD=${CDP_PULL_REQUEST_NUMBER+"true"}
-    if [[ ${CDP_TARGET_BRANCH} == "master" && ${IS_PR_BUILD} != "true" ]]
-    then
-      docker push $image
+    if [[ $CDP_TARGET_BRANCH == master && ! $CDP_PULL_REQUEST_NUMBER ]]; then
+      RELEASE_VERSION=$(git describe --tags --always --dirty)
+      IMAGE=registry-write.opensource.zalan.do/teapot/external-dns:${RELEASE_VERSION}
+    else
+      IMAGE=registry-write.opensource.zalan.do/teapot/external-dns-test:${CDP_BUILD_VERSION}
     fi
+    docker build --squash --tag "$IMAGE" .
+    docker push "$IMAGE"

pkg/apis/externaldns/types.go

@@ -123,7 +123,7 @@ var defaultConfig = &Config{
 	Registry:                 "txt",
 	TXTOwnerID:               "default",
 	TXTPrefix:                "",
-	TXTCacheInterval:         time.Hour,
+	TXTCacheInterval:         0,
 	Interval:                 time.Minute,
 	Once:                     false,
 	DryRun:                   false,
@@ -223,7 +223,7 @@ func (cfg *Config) ParseFlags(args []string) error {
 	app.Flag("txt-prefix", "When using the TXT registry, a custom string that's prefixed to each ownership DNS record (optional)").Default(defaultConfig.TXTPrefix).StringVar(&cfg.TXTPrefix)
 
 	// Flags related to the main control loop
-	app.Flag("txt-cache-interval", "The interval between cache synchronizations in duration format (default: 1h)").Default(defaultConfig.TXTCacheInterval.String()).DurationVar(&cfg.TXTCacheInterval)
+	app.Flag("txt-cache-interval", "The interval between cache synchronizations in duration format (default: disabled)").Default(defaultConfig.TXTCacheInterval.String()).DurationVar(&cfg.TXTCacheInterval)
 	app.Flag("interval", "The interval between two consecutive synchronizations in duration format (default: 1m)").Default(defaultConfig.Interval.String()).DurationVar(&cfg.Interval)
 	app.Flag("once", "When enabled, exits the synchronization loop after the first iteration (default: disabled)").BoolVar(&cfg.Once)
 	app.Flag("dry-run", "When enabled, prints DNS record changes rather than actually performing them (default: disabled)").BoolVar(&cfg.DryRun)

pkg/apis/externaldns/types_test.go

@@ -58,7 +58,7 @@ var (
 		Registry:              "txt",
 		TXTOwnerID:            "default",
 		TXTPrefix:             "",
-		TXTCacheInterval:      time.Hour,
+		TXTCacheInterval:      0,
 		Interval:              time.Minute,
 		Once:                  false,
 		DryRun:                false,

provider/google.go

@@ -143,10 +143,10 @@ func NewGoogleProvider(project string, domainFilter DomainFilter, zoneIDFilter Z
 	}
 
 	provider := &GoogleProvider{
-		project:      project,
-		domainFilter: domainFilter,
-		zoneIDFilter: zoneIDFilter,
-		dryRun:       dryRun,
+		project:                  project,
+		domainFilter:             domainFilter,
+		zoneIDFilter:             zoneIDFilter,
+		dryRun:                   dryRun,
 		resourceRecordSetsClient: resourceRecordSetsService{dnsClient.ResourceRecordSets},
 		managedZonesClient:       managedZonesService{dnsClient.ManagedZones},
 		changesClient:            changesService{dnsClient.Changes},

provider/google_test.go

@@ -569,10 +569,10 @@ func validateChangeRecord(t *testing.T, record *dns.ResourceRecordSet, expected
 
 func newGoogleProvider(t *testing.T, domainFilter DomainFilter, zoneIDFilter ZoneIDFilter, dryRun bool, records []*endpoint.Endpoint) *GoogleProvider {
 	provider := &GoogleProvider{
-		project:      "zalando-external-dns-test",
-		domainFilter: domainFilter,
-		zoneIDFilter: zoneIDFilter,
-		dryRun:       false,
+		project:                  "zalando-external-dns-test",
+		domainFilter:             domainFilter,
+		zoneIDFilter:             zoneIDFilter,
+		dryRun:                   false,
 		resourceRecordSetsClient: &mockResourceRecordSetsClient{},
 		managedZonesClient:       &mockManagedZonesClient{},
 		changesClient:            &mockChangesClient{},

registry/txt.go

@@ -107,8 +107,10 @@ func (im *TXTRegistry) Records() ([]*endpoint.Endpoint, error) {
 	}
 
 	// Update the cache.
-	im.recordsCache = endpoints
-	im.recordsCacheRefreshTime = time.Now()
+	if im.cacheInterval > 0 {
+		im.recordsCache = endpoints
+		im.recordsCacheRefreshTime = time.Now()
+	}
 
 	return endpoints, nil
 }
@@ -127,9 +129,8 @@ func (im *TXTRegistry) ApplyChanges(changes *plan.Changes) error {
 		txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), endpoint.RecordTypeTXT, r.Labels.Serialize(true))
 		filteredChanges.Create = append(filteredChanges.Create, txt)
 
-		// Add to the cache.
-		if im.recordsCache != nil {
-			im.recordsCache = append(im.recordsCache, txt)
+		if im.cacheInterval > 0 {
+			im.addToCache(r)
 		}
 	}
 
@@ -140,27 +141,31 @@ func (im *TXTRegistry) ApplyChanges(changes *plan.Changes) error {
 		// !!! TXT record value is uniquely generated from the Labels of the endpoint. Hence old TXT record can be uniquely reconstructed
 		filteredChanges.Delete = append(filteredChanges.Delete, txt)
 
-		// Remove from the cache.
-		im.removeFromCache(txt)
+		if im.cacheInterval > 0 {
+			im.removeFromCache(r)
+		}
 	}
 
-	// make sure TXT records are consistently updated as well
-	for _, r := range filteredChanges.UpdateNew {
-		txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), endpoint.RecordTypeTXT, r.Labels.Serialize(true))
-		filteredChanges.UpdateNew = append(filteredChanges.UpdateNew, txt)
-
-		// Update the cache.
-		im.updateCache(txt)
-	}
 	// make sure TXT records are consistently updated as well
 	for _, r := range filteredChanges.UpdateOld {
 		txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), endpoint.RecordTypeTXT, r.Labels.Serialize(true))
 		// when we updateOld TXT records for which value has changed (due to new label) this would still work because
 		// !!! TXT record value is uniquely generated from the Labels of the endpoint. Hence old TXT record can be uniquely reconstructed
 		filteredChanges.UpdateOld = append(filteredChanges.UpdateOld, txt)
+		// remove old version of record from cache
+		if im.cacheInterval > 0 {
+			im.removeFromCache(r)
+		}
+	}
 
-		// Update the cache.
-		im.updateCache(txt)
+	// make sure TXT records are consistently updated as well
+	for _, r := range filteredChanges.UpdateNew {
+		txt := endpoint.NewEndpoint(im.mapper.toTXTName(r.DNSName), endpoint.RecordTypeTXT, r.Labels.Serialize(true))
+		filteredChanges.UpdateNew = append(filteredChanges.UpdateNew, txt)
+		// add new version of record to cache
+		if im.cacheInterval > 0 {
+			im.addToCache(r)
+		}
 	}
 
 	return im.provider.ApplyChanges(filteredChanges)
@@ -201,35 +206,23 @@ func (pr prefixNameMapper) toTXTName(endpointDNSName string) string {
 	return pr.prefix + endpointDNSName
 }
 
-func (im *TXTRegistry) removeFromCache(txt *endpoint.Endpoint) {
-	if im.recordsCache == nil || txt == nil {
-		// return early.
-		return
-	}
-
-	for i, e := range im.recordsCache {
-		if e.DNSName == txt.DNSName && e.RecordType == txt.RecordType {
-			// We found a match delete the endpoint from the cache.
-			im.recordsCache = append(im.recordsCache[:i], im.recordsCache[i+1:]...)
-			return
-		}
+func (im *TXTRegistry) addToCache(ep *endpoint.Endpoint) {
+	if im.recordsCache != nil {
+		im.recordsCache = append(im.recordsCache, ep)
 	}
 }
 
-func (im *TXTRegistry) updateCache(txt *endpoint.Endpoint) {
-	if im.recordsCache == nil || txt == nil {
+func (im *TXTRegistry) removeFromCache(ep *endpoint.Endpoint) {
+	if im.recordsCache == nil || ep == nil {
 		// return early.
 		return
 	}
 
 	for i, e := range im.recordsCache {
-		if e.DNSName == txt.DNSName && e.RecordType == txt.RecordType {
-			// We found a match update the endpoint in the cache.
-			im.recordsCache[i] = txt
+		if e.DNSName == ep.DNSName && e.RecordType == ep.RecordType && e.Targets.Same(ep.Targets) {
+			// We found a match delete the endpoint from the cache.
+			im.recordsCache = append(im.recordsCache[:i], im.recordsCache[i+1:]...)
 			return
 		}
 	}
-
-	// We couldn't find a match so let's just add it to the cache.
-	im.recordsCache = append(im.recordsCache, txt)
 }

registry/txt_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package registry
 
 import (
+	"reflect"
 	"testing"
 	"time"
 
@@ -361,39 +362,51 @@ func TestCacheMethods(t *testing.T) {
 		cacheInterval: time.Hour,
 	}
 
-	// test updating a record.
-	registry.updateCache(newEndpointWithOwner("thing.com", "1.2.3.6", "A", "owner2"))
-	found := false
-	// ensure it was updated
-	for _, e := range registry.recordsCache {
-		if e.DNSName == "thing.com" && e.RecordType == "A" {
-			t.Logf("targets: %#v", e.Targets)
-			if e.Targets.Same([]string{"1.2.3.6"}) {
-				found = true
-				break
-			}
-		}
+	expectedCacheAfterAdd := []*endpoint.Endpoint{
+		newEndpointWithOwner("thing.com", "1.2.3.4", "A", "owner"),
+		newEndpointWithOwner("thing1.com", "1.2.3.6", "A", "owner"),
+		newEndpointWithOwner("thing2.com", "1.2.3.4", "CNAME", "owner"),
+		newEndpointWithOwner("thing3.com", "1.2.3.4", "A", "owner"),
+		newEndpointWithOwner("thing4.com", "1.2.3.4", "A", "owner"),
+		newEndpointWithOwner("thing5.com", "1.2.3.5", "A", "owner"),
 	}
 
-	if !found {
-		t.Fatal("could not find updated record in cache")
+	expectedCacheAfterUpdate := []*endpoint.Endpoint{
+		newEndpointWithOwner("thing1.com", "1.2.3.6", "A", "owner"),
+		newEndpointWithOwner("thing2.com", "1.2.3.4", "CNAME", "owner"),
+		newEndpointWithOwner("thing3.com", "1.2.3.4", "A", "owner"),
+		newEndpointWithOwner("thing4.com", "1.2.3.4", "A", "owner"),
+		newEndpointWithOwner("thing5.com", "1.2.3.5", "A", "owner"),
+		newEndpointWithOwner("thing.com", "1.2.3.6", "A", "owner2"),
+	}
+
+	expectedCacheAfterDelete := []*endpoint.Endpoint{
+		newEndpointWithOwner("thing1.com", "1.2.3.6", "A", "owner"),
+		newEndpointWithOwner("thing2.com", "1.2.3.4", "CNAME", "owner"),
+		newEndpointWithOwner("thing3.com", "1.2.3.4", "A", "owner"),
+		newEndpointWithOwner("thing4.com", "1.2.3.4", "A", "owner"),
+		newEndpointWithOwner("thing5.com", "1.2.3.5", "A", "owner"),
+	}
+	// test add cache
+	registry.addToCache(newEndpointWithOwner("thing5.com", "1.2.3.5", "A", "owner"))
+
+	if !reflect.DeepEqual(expectedCacheAfterAdd, registry.recordsCache) {
+		t.Fatalf("expected endpoints should match endpoints from cache: expected %v, but got %v", expectedCacheAfterAdd, registry.recordsCache)
+	}
+
+	// test update cache
+	registry.removeFromCache(newEndpointWithOwner("thing.com", "1.2.3.4", "A", "owner"))
+	registry.addToCache(newEndpointWithOwner("thing.com", "1.2.3.6", "A", "owner2"))
+	// ensure it was updated
+	if !reflect.DeepEqual(expectedCacheAfterUpdate, registry.recordsCache) {
+		t.Fatalf("expected endpoints should match endpoints from cache: expected %v, but got %v", expectedCacheAfterUpdate, registry.recordsCache)
 	}
 
 	// test deleting a record
 	registry.removeFromCache(newEndpointWithOwner("thing.com", "1.2.3.6", "A", "owner2"))
 	// ensure it was deleted
-	found = false
-	for _, e := range registry.recordsCache {
-		if e.DNSName == "thing.com" && e.RecordType == "A" {
-			if e.Targets.Same([]string{"1.2.3.6"}) {
-				found = true
-				break
-			}
-		}
-	}
-
-	if found {
-		t.Fatal("should not have been able to find record after deleting")
+	if !reflect.DeepEqual(expectedCacheAfterDelete, registry.recordsCache) {
+		t.Fatalf("expected endpoints should match endpoints from cache: expected %v, but got %v", expectedCacheAfterDelete, registry.recordsCache)
 	}
 }