Merge pull request #39957 from justinsb/dnsprovider_upsert

Automatic merge from submit-queue

dnsprovider: Add upsert

Although Google Cloud DNS requires strict add & remove calls, most
dnsproviders actually support upsert, and an add & remove is much more
expensive (primarily because of the need to fetch the pre-image).

Add support for 'upsert' operations, which don't require the pre-image,
and simply overwrite the existing record.  This is much cheaper on
Amazon Route53, for example.

```release-note
NONE
```

federation/pkg/dnsprovider/dns.go

@@ -68,6 +68,10 @@ type ResourceRecordChangeset interface {
 	// Remove adds the removal of a ResourceRecordSet in the Zone to the changeset
 	// The supplied ResourceRecordSet must match one of the existing recordsets (obtained via List()) exactly.
 	Remove(ResourceRecordSet) ResourceRecordChangeset
+	// Upsert adds an "create or update" operation for the ResourceRecordSet in the Zone to the changeset
+	// Note: the implementation may translate this into a Remove followed by an Add operation.
+	// If you have the pre-image, it will likely be more efficient to call Remove and Add.
+	Upsert(ResourceRecordSet) ResourceRecordChangeset
 	// Apply applies the accumulated operations to the Zone.
 	Apply() error
 	// IsEmpty returns true if there are no accumulated operations.

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

@@ -31,6 +31,7 @@ type ResourceRecordChangeset struct {
 
 	additions []dnsprovider.ResourceRecordSet
 	removals  []dnsprovider.ResourceRecordSet
+	upserts   []dnsprovider.ResourceRecordSet
 }
 
 func (c *ResourceRecordChangeset) Add(rrset dnsprovider.ResourceRecordSet) dnsprovider.ResourceRecordChangeset {
@@ -43,6 +44,11 @@ func (c *ResourceRecordChangeset) Remove(rrset dnsprovider.ResourceRecordSet) dn
 	return c
 }
 
+func (c *ResourceRecordChangeset) Upsert(rrset dnsprovider.ResourceRecordSet) dnsprovider.ResourceRecordChangeset {
+	c.upserts = append(c.upserts, rrset)
+	return c
+}
+
 // buildChange converts a dnsprovider.ResourceRecordSet to a route53.Change request
 func buildChange(action string, rrs dnsprovider.ResourceRecordSet) *route53.Change {
 	change := &route53.Change{
@@ -78,6 +84,11 @@ func (c *ResourceRecordChangeset) Apply() error {
 		changes = append(changes, change)
 	}
 
+	for _, upsert := range c.upserts {
+		change := buildChange(route53.ChangeActionUpsert, upsert)
+		changes = append(changes, change)
+	}
+
 	if len(changes) == 0 {
 		return nil
 	}

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

@@ -35,6 +35,7 @@ type ChangeSetType string
 const (
 	ADDITION = ChangeSetType("ADDITION")
 	DELETION = ChangeSetType("DELETION")
+	UPSERT   = ChangeSetType("UPSERT")
 )
 
 type ChangeSet struct {
@@ -63,6 +64,11 @@ func (c *ResourceRecordChangeset) IsEmpty() bool {
 	return len(c.changeset) == 0
 }
 
+func (c *ResourceRecordChangeset) Upsert(rrset dnsprovider.ResourceRecordSet) dnsprovider.ResourceRecordChangeset {
+	c.changeset = append(c.changeset, ChangeSet{cstype: UPSERT, rrset: rrset})
+	return c
+}
+
 func (c *ResourceRecordChangeset) Apply() error {
 	ctx := context.Background()
 	etcdPathPrefix := c.zone.zones.intf.etcdPathPrefix
@@ -74,7 +80,11 @@ func (c *ResourceRecordChangeset) Apply() error {
 
 	for _, changeset := range c.changeset {
 		switch changeset.cstype {
-		case ADDITION:
+		case ADDITION, UPSERT:
+			checkNotExists := changeset.cstype == ADDITION
+
+			// TODO: I think the semantics of the other providers are different; they operate at the record level, not the individual rrdata level
+			// In other words: we should insert/replace all the records for the key
 			for _, rrdata := range changeset.rrset.Rrdatas() {
 				b, err := json.Marshal(&dnsmsg.Service{Host: rrdata, TTL: uint32(changeset.rrset.Ttl()), Group: changeset.rrset.Name()})
 				if err != nil {
@@ -84,9 +94,11 @@ func (c *ResourceRecordChangeset) Apply() error {
 				recordLabel := getHash(rrdata)
 				recordKey := buildDNSNameString(changeset.rrset.Name(), recordLabel)
 
-				response, err := c.zone.zones.intf.etcdKeysAPI.Get(ctx, dnsmsg.Path(recordKey, etcdPathPrefix), getOpts)
-				if err == nil && response != nil {
-					return fmt.Errorf("Key already exist, key: %v", recordKey)
+				if checkNotExists {
+					response, err := c.zone.zones.intf.etcdKeysAPI.Get(ctx, dnsmsg.Path(recordKey, etcdPathPrefix), getOpts)
+					if err == nil && response != nil {
+						return fmt.Errorf("Key already exist, key: %v", recordKey)
+					}
 				}
 
 				_, err = c.zone.zones.intf.etcdKeysAPI.Set(ctx, dnsmsg.Path(recordKey, etcdPathPrefix), recordValue, setOpts)
@@ -94,7 +106,10 @@ func (c *ResourceRecordChangeset) Apply() error {
 					return err
 				}
 			}
+
 		case DELETION:
+			// TODO: I think the semantics of the other providers are different; they operate at the record level, not the individual rrdata level
+			// In other words: we should delete all the records for the key, only if it matches exactly
 			for _, rrdata := range changeset.rrset.Rrdatas() {
 				recordLabel := getHash(rrdata)
 				recordKey := buildDNSNameString(changeset.rrset.Name(), recordLabel)

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

@@ -31,6 +31,7 @@ type ResourceRecordChangeset struct {
 
 	additions []dnsprovider.ResourceRecordSet
 	removals  []dnsprovider.ResourceRecordSet
+	upserts   []dnsprovider.ResourceRecordSet
 }
 
 func (c *ResourceRecordChangeset) Add(rrset dnsprovider.ResourceRecordSet) dnsprovider.ResourceRecordChangeset {
@@ -43,10 +44,16 @@ func (c *ResourceRecordChangeset) Remove(rrset dnsprovider.ResourceRecordSet) dn
 	return c
 }
 
+func (c *ResourceRecordChangeset) Upsert(rrset dnsprovider.ResourceRecordSet) dnsprovider.ResourceRecordChangeset {
+	c.upserts = append(c.upserts, rrset)
+	return c
+}
+
 func (c *ResourceRecordChangeset) Apply() error {
 	rrsets := c.rrsets
 
 	service := rrsets.zone.zones.interface_.service.Changes()
+
 	var additions []interfaces.ResourceRecordSet
 	for _, r := range c.additions {
 		additions = append(additions, r.(ResourceRecordSet).impl)
@@ -56,6 +63,40 @@ func (c *ResourceRecordChangeset) Apply() error {
 		deletions = append(deletions, r.(ResourceRecordSet).impl)
 	}
 
+	if len(c.upserts) != 0 {
+		// TODO: We could maybe tweak this to fetch just the records we care about
+		// although not clear when this would be a win.  N=1 obviously so though...
+		before, err := c.rrsets.List()
+		if err != nil {
+			return fmt.Errorf("error fetching recordset images for upsert operation: %v", err)
+		}
+
+		upsertMap := make(map[string]dnsprovider.ResourceRecordSet)
+		for _, upsert := range c.upserts {
+			key := string(upsert.Type()) + "::" + upsert.Name()
+			upsertMap[key] = upsert
+		}
+
+		for _, b := range before {
+			key := string(b.Type()) + "::" + b.Name()
+			upsert := upsertMap[key]
+			if upsert == nil {
+				continue
+			}
+
+			deletions = append(deletions, b.(ResourceRecordSet).impl)
+			additions = append(additions, upsert.(ResourceRecordSet).impl)
+
+			// Mark as seen
+			delete(upsertMap, key)
+		}
+
+		// Anything left in the map must be an addition
+		for _, upsert := range upsertMap {
+			additions = append(additions, upsert.(ResourceRecordSet).impl)
+		}
+	}
+
 	change := service.NewChange(additions, deletions)
 	newChange, err := service.Create(rrsets.project(), rrsets.zone.impl.Name(), change).Do()
 	if err != nil {