Validate AWS record values size during batch set generation

main.go

@@ -232,6 +232,8 @@ func main() {
 				ZoneTypeFilter:        zoneTypeFilter,
 				ZoneTagFilter:         zoneTagFilter,
 				BatchChangeSize:       cfg.AWSBatchChangeSize,
+				BatchChangeSizeBytes:  cfg.AWSBatchChangeSizeBytes,
+				BatchChangeSizeValues: cfg.AWSBatchChangeSizeValues,
 				BatchChangeInterval:   cfg.AWSBatchChangeInterval,
 				EvaluateTargetHealth:  cfg.AWSEvaluateTargetHealth,
 				PreferCNAME:           cfg.AWSPreferCNAME,

pkg/apis/externaldns/types.go

@@ -86,6 +86,8 @@ type Config struct {
 	AWSAssumeRole                      string
 	AWSAssumeRoleExternalID            string
 	AWSBatchChangeSize                 int
+	AWSBatchChangeSizeBytes            int
+	AWSBatchChangeSizeValues           int
 	AWSBatchChangeInterval             time.Duration
 	AWSEvaluateTargetHealth            bool
 	AWSAPIRetries                      int
@@ -258,6 +260,8 @@ var defaultConfig = &Config{
 	AWSAssumeRole:               "",
 	AWSAssumeRoleExternalID:     "",
 	AWSBatchChangeSize:          1000,
+	AWSBatchChangeSizeBytes:     32000,
+	AWSBatchChangeSizeValues:    1000,
 	AWSBatchChangeInterval:      time.Second,
 	AWSEvaluateTargetHealth:     true,
 	AWSAPIRetries:               3,
@@ -477,6 +481,8 @@ func (cfg *Config) ParseFlags(args []string) error {
 	app.Flag("aws-assume-role", "When using the AWS API, assume this IAM role. Useful for hosted zones in another AWS account. Specify the full ARN, e.g. `arn:aws:iam::123455567:role/external-dns` (optional)").Default(defaultConfig.AWSAssumeRole).StringVar(&cfg.AWSAssumeRole)
 	app.Flag("aws-assume-role-external-id", "When using the AWS API and assuming a role then specify this external ID` (optional)").Default(defaultConfig.AWSAssumeRoleExternalID).StringVar(&cfg.AWSAssumeRoleExternalID)
 	app.Flag("aws-batch-change-size", "When using the AWS provider, set the maximum number of changes that will be applied in each batch.").Default(strconv.Itoa(defaultConfig.AWSBatchChangeSize)).IntVar(&cfg.AWSBatchChangeSize)
+	app.Flag("aws-batch-change-size-bytes", "When using the AWS provider, set the maximum byte size that will be applied in each batch.").Default(strconv.Itoa(defaultConfig.AWSBatchChangeSizeBytes)).IntVar(&cfg.AWSBatchChangeSizeBytes)
+	app.Flag("aws-batch-change-size-values", "When using the AWS provider, set the maximum total record values that will be applied in each batch.").Default(strconv.Itoa(defaultConfig.AWSBatchChangeSizeValues)).IntVar(&cfg.AWSBatchChangeSizeValues)
 	app.Flag("aws-batch-change-interval", "When using the AWS provider, set the interval between batch changes.").Default(defaultConfig.AWSBatchChangeInterval.String()).DurationVar(&cfg.AWSBatchChangeInterval)
 	app.Flag("aws-evaluate-target-health", "When using the AWS provider, set whether to evaluate the health of a DNS target (default: enabled, disable with --no-aws-evaluate-target-health)").Default(strconv.FormatBool(defaultConfig.AWSEvaluateTargetHealth)).BoolVar(&cfg.AWSEvaluateTargetHealth)
 	app.Flag("aws-api-retries", "When using the AWS API, set the maximum number of retries before giving up.").Default(strconv.Itoa(defaultConfig.AWSAPIRetries)).IntVar(&cfg.AWSAPIRetries)

pkg/apis/externaldns/types_test.go

@@ -58,6 +58,8 @@ var (
 		AWSAssumeRole:               "",
 		AWSAssumeRoleExternalID:     "",
 		AWSBatchChangeSize:          1000,
+		AWSBatchChangeSizeBytes:     32000,
+		AWSBatchChangeSizeValues:    1000,
 		AWSBatchChangeInterval:      time.Second,
 		AWSEvaluateTargetHealth:     true,
 		AWSAPIRetries:               3,
@@ -169,6 +171,8 @@ var (
 		AWSAssumeRole:               "some-other-role",
 		AWSAssumeRoleExternalID:     "pg2000",
 		AWSBatchChangeSize:          100,
+		AWSBatchChangeSizeBytes:     16000,
+		AWSBatchChangeSizeValues:    100,
 		AWSBatchChangeInterval:      time.Second * 2,
 		AWSEvaluateTargetHealth:     false,
 		AWSAPIRetries:               13,
@@ -354,6 +358,8 @@ func TestParseFlags(t *testing.T) {
 				"--aws-assume-role=some-other-role",
 				"--aws-assume-role-external-id=pg2000",
 				"--aws-batch-change-size=100",
+				"--aws-batch-change-size-bytes=16000",
+				"--aws-batch-change-size-values=100",
 				"--aws-batch-change-interval=2s",
 				"--aws-api-retries=13",
 				"--aws-prefer-cname",
@@ -476,6 +482,8 @@ func TestParseFlags(t *testing.T) {
 				"EXTERNAL_DNS_AWS_ASSUME_ROLE":                 "some-other-role",
 				"EXTERNAL_DNS_AWS_ASSUME_ROLE_EXTERNAL_ID":     "pg2000",
 				"EXTERNAL_DNS_AWS_BATCH_CHANGE_SIZE":           "100",
+				"EXTERNAL_DNS_AWS_BATCH_CHANGE_SIZE_BYTES":     "16000",
+				"EXTERNAL_DNS_AWS_BATCH_CHANGE_SIZE_VALUES":    "100",
 				"EXTERNAL_DNS_AWS_BATCH_CHANGE_INTERVAL":       "2s",
 				"EXTERNAL_DNS_AWS_EVALUATE_TARGET_HEALTH":      "0",
 				"EXTERNAL_DNS_AWS_API_RETRIES":                 "13",

provider/aws/aws.go

@@ -206,6 +206,8 @@ type Route53API interface {
 type Route53Change struct {
 	route53.Change
 	OwnedRecord string
+	sizeBytes   int
+	sizeValues  int
 }
 
 type Route53Changes []*Route53Change
@@ -230,6 +232,8 @@ type AWSProvider struct {
 	client                Route53API
 	dryRun                bool
 	batchChangeSize       int
+	batchChangeSizeBytes  int
+	batchChangeSizeValues int
 	batchChangeInterval   time.Duration
 	evaluateTargetHealth  bool
 	// only consider hosted zones managing domains ending in this suffix
@@ -253,6 +257,8 @@ type AWSConfig struct {
 	ZoneTypeFilter        provider.ZoneTypeFilter
 	ZoneTagFilter         provider.ZoneTagFilter
 	BatchChangeSize       int
+	BatchChangeSizeBytes  int
+	BatchChangeSizeValues int
 	BatchChangeInterval   time.Duration
 	EvaluateTargetHealth  bool
 	PreferCNAME           bool
@@ -269,6 +275,8 @@ func NewAWSProvider(awsConfig AWSConfig, client Route53API) (*AWSProvider, error
 		zoneTypeFilter:        awsConfig.ZoneTypeFilter,
 		zoneTagFilter:         awsConfig.ZoneTagFilter,
 		batchChangeSize:       awsConfig.BatchChangeSize,
+		batchChangeSizeBytes:  awsConfig.BatchChangeSizeBytes,
+		batchChangeSizeValues: awsConfig.BatchChangeSizeValues,
 		batchChangeInterval:   awsConfig.BatchChangeInterval,
 		evaluateTargetHealth:  awsConfig.EvaluateTargetHealth,
 		preferCNAME:           awsConfig.PreferCNAME,
@@ -565,7 +573,8 @@ func (p *AWSProvider) submitChanges(ctx context.Context, changes Route53Changes,
 		retriedChanges, newChanges := findChangesInQueue(cs, p.failedChangesQueue[z])
 		p.failedChangesQueue[z] = nil
 
-		batchCs := append(batchChangeSet(newChanges, p.batchChangeSize), batchChangeSet(retriedChanges, p.batchChangeSize)...)
+		batchCs := append(batchChangeSet(newChanges, p.batchChangeSize, p.batchChangeSizeBytes, p.batchChangeSizeValues),
+			batchChangeSet(retriedChanges, p.batchChangeSize, p.batchChangeSizeBytes, p.batchChangeSizeValues)...)
 		for i, b := range batchCs {
 			if len(b) == 0 {
 				continue
@@ -719,6 +728,8 @@ func (p *AWSProvider) newChange(action string, ep *endpoint.Endpoint) (*Route53C
 				Name: aws.String(ep.DNSName),
 			},
 		},
+		sizeBytes:  0,
+		sizeValues: 0,
 	}
 	dualstack := false
 	if targetHostedZone := isAWSAlias(ep); targetHostedZone != "" {
@@ -736,6 +747,8 @@ func (p *AWSProvider) newChange(action string, ep *endpoint.Endpoint) (*Route53C
 			HostedZoneId:         aws.String(cleanZoneID(targetHostedZone)),
 			EvaluateTargetHealth: aws.Bool(evalTargetHealth),
 		}
+		change.sizeBytes += len([]byte(ep.Targets[0]))
+		change.sizeValues += 1
 	} else {
 		change.ResourceRecordSet.Type = aws.String(ep.RecordType)
 		if !ep.RecordTTL.IsConfigured() {
@@ -748,9 +761,18 @@ func (p *AWSProvider) newChange(action string, ep *endpoint.Endpoint) (*Route53C
 			change.ResourceRecordSet.ResourceRecords[idx] = &route53.ResourceRecord{
 				Value: aws.String(val),
 			}
+			change.sizeBytes += len([]byte(val))
+			change.sizeValues += 1
 		}
 	}
 
+	if action == route53.ChangeActionUpsert {
+		// When the value of the Action element is UPSERT, each ResourceRecord element and each character in a Value
+		// element is counted twice
+		change.sizeBytes *= 2
+		change.sizeValues *= 2
+	}
+
 	setIdentifier := ep.SetIdentifier
 	if setIdentifier != "" {
 		change.ResourceRecordSet.SetIdentifier = aws.String(setIdentifier)
@@ -856,8 +878,26 @@ func (p *AWSProvider) tagsForZone(ctx context.Context, zoneID string) (map[strin
 	return tagMap, nil
 }
 
-func batchChangeSet(cs Route53Changes, batchSize int) []Route53Changes {
+// count bytes for all changes values
-	if len(cs) <= batchSize {
+func countChangeBytes(cs Route53Changes) int {
+	count := 0
+	for _, c := range cs {
+		count += c.sizeBytes
+	}
+	return count
+}
+
+// count total value count for all changes
+func countChangeValues(cs Route53Changes) int {
+	count := 0
+	for _, c := range cs {
+		count += c.sizeValues
+	}
+	return count
+}
+
+func batchChangeSet(cs Route53Changes, batchSize int, batchSizeBytes int, batchSizeValues int) []Route53Changes {
+	if len(cs) <= batchSize && countChangeBytes(cs) <= batchSizeBytes && countChangeValues(cs) <= batchSizeValues {
 		res := sortChangesByActionNameType(cs)
 		return []Route53Changes{res}
 	}
@@ -880,7 +920,10 @@ func batchChangeSet(cs Route53Changes, batchSize int) []Route53Changes {
 			continue
 		}
 
-		if len(currentBatch)+len(v) > batchSize {
+		bytes := countChangeBytes(currentBatch) + countChangeBytes(v)
+		values := countChangeValues(currentBatch) + countChangeValues(v)
+
+		if len(currentBatch)+len(v) > batchSize || bytes > batchSizeBytes || values > batchSizeValues {
 			// currentBatch would be too large if we add this changeset;
 			// add currentBatch to batchChanges and start a new currentBatch
 			batchChanges = append(batchChanges, sortChangesByActionNameType(currentBatch))

provider/aws/aws_test.go

@@ -19,6 +19,7 @@ package aws
 import (
 	"context"
 	"fmt"
+	"math"
 	"net"
 	"sort"
 	"strings"
@@ -40,6 +41,8 @@ import (
 
 const (
 	defaultBatchChangeSize       = 4000
+	defaultBatchChangeSizeBytes  = 32000
+	defaultBatchChangeSizeValues = 1000
 	defaultBatchChangeInterval   = time.Second
 	defaultEvaluateTargetHealth  = true
 )
@@ -1527,7 +1530,7 @@ func TestAWSBatchChangeSet(t *testing.T) {
 		})
 	}
 
-	batchCs := batchChangeSet(cs, defaultBatchChangeSize)
+	batchCs := batchChangeSet(cs, defaultBatchChangeSize, defaultBatchChangeSizeBytes, defaultBatchChangeSizeValues)
 
 	require.Equal(t, 1, len(batchCs))
 
@@ -1565,7 +1568,7 @@ func TestAWSBatchChangeSetExceeding(t *testing.T) {
 		)
 	}
 
-	batchCs := batchChangeSet(cs, testLimit)
+	batchCs := batchChangeSet(cs, testLimit, defaultBatchChangeSizeBytes, defaultBatchChangeSizeValues)
 
 	require.Equal(t, expectedBatchCount, len(batchCs))
 
@@ -1603,11 +1606,227 @@ func TestAWSBatchChangeSetExceedingNameChange(t *testing.T) {
 		)
 	}
 
-	batchCs := batchChangeSet(cs, testLimit)
+	batchCs := batchChangeSet(cs, testLimit, defaultBatchChangeSizeBytes, defaultBatchChangeSizeValues)
 
 	require.Equal(t, 0, len(batchCs))
 }
 
+func TestAWSBatchChangeSetExceedingBytesLimit(t *testing.T) {
+	var cs Route53Changes
+	const testCount = 50
+	const testLimit = 100
+	const groupSize = 2
+	// Bytes for each name
+	var testBytes = len([]byte("1.2.3.4")) + len([]byte("test-record"))
+	// testCount / groupSize / (testLimit // bytes)
+	var expectedBatchCountFloat = float64(testCount) / float64(groupSize) / float64(testLimit/testBytes)
+	// Round up
+	var expectedBatchCount = int(math.Floor(expectedBatchCountFloat + 0.5))
+
+	for i := 1; i <= testCount; i += groupSize {
+		cs = append(cs,
+			&Route53Change{
+				Change: route53.Change{
+					Action: aws.String(route53.ChangeActionCreate),
+					ResourceRecordSet: &route53.ResourceRecordSet{
+						Name: aws.String(fmt.Sprintf("host-%d", i)),
+						Type: aws.String("A"),
+						ResourceRecords: []*route53.ResourceRecord{
+							{
+								Value: aws.String("1.2.3.4"),
+							},
+						},
+					},
+				},
+				sizeBytes:  len([]byte("1.2.3.4")),
+				sizeValues: 1,
+			},
+			&Route53Change{
+				Change: route53.Change{
+					Action: aws.String(route53.ChangeActionCreate),
+					ResourceRecordSet: &route53.ResourceRecordSet{
+						Name: aws.String(fmt.Sprintf("host-%d", i)),
+						Type: aws.String("TXT"),
+						ResourceRecords: []*route53.ResourceRecord{
+							{
+								Value: aws.String("txt-record"),
+							},
+						},
+					},
+				},
+				sizeBytes:  len([]byte("txt-record")),
+				sizeValues: 1,
+			},
+		)
+	}
+
+	batchCs := batchChangeSet(cs, defaultBatchChangeSize, testLimit, defaultBatchChangeSizeValues)
+
+	require.Equal(t, expectedBatchCount, len(batchCs))
+}
+
+func TestAWSBatchChangeSetExceedingBytesLimitUpsert(t *testing.T) {
+	var cs Route53Changes
+	const testCount = 50
+	const testLimit = 100
+	const groupSize = 2
+	// Bytes for each name multiplied by 2 for Upsert records
+	var testBytes = (len([]byte("1.2.3.4")) + len([]byte("test-record"))) * 2
+	// testCount / groupSize / (testLimit // bytes)
+	var expectedBatchCountFloat = float64(testCount) / float64(groupSize) / float64(testLimit/testBytes)
+	// Round up
+	var expectedBatchCount = int(math.Floor(expectedBatchCountFloat + 0.5))
+
+	for i := 1; i <= testCount; i += groupSize {
+		cs = append(cs,
+			&Route53Change{
+				Change: route53.Change{
+					Action: aws.String(route53.ChangeActionUpsert),
+					ResourceRecordSet: &route53.ResourceRecordSet{
+						Name: aws.String(fmt.Sprintf("host-%d", i)),
+						Type: aws.String("A"),
+						ResourceRecords: []*route53.ResourceRecord{
+							{
+								Value: aws.String("1.2.3.4"),
+							},
+						},
+					},
+				},
+				sizeBytes:  len([]byte("1.2.3.4")) * 2,
+				sizeValues: 1,
+			},
+			&Route53Change{
+				Change: route53.Change{
+					Action: aws.String(route53.ChangeActionUpsert),
+					ResourceRecordSet: &route53.ResourceRecordSet{
+						Name: aws.String(fmt.Sprintf("host-%d", i)),
+						Type: aws.String("TXT"),
+						ResourceRecords: []*route53.ResourceRecord{
+							{
+								Value: aws.String("txt-record"),
+							},
+						},
+					},
+				},
+				sizeBytes:  len([]byte("txt-record")) * 2,
+				sizeValues: 1,
+			},
+		)
+	}
+
+	batchCs := batchChangeSet(cs, defaultBatchChangeSize, testLimit, defaultBatchChangeSizeValues)
+
+	require.Equal(t, expectedBatchCount, len(batchCs))
+}
+
+func TestAWSBatchChangeSetExceedingValuesLimit(t *testing.T) {
+	var cs Route53Changes
+	const testCount = 50
+	const testLimit = 100
+	const groupSize = 2
+	// Values for each group
+	const testValues = 2
+	// testCount / groupSize / (testLimit // bytes)
+	var expectedBatchCountFloat = float64(testCount) / float64(groupSize) / float64(testLimit/testValues)
+	// Round up
+	var expectedBatchCount = int(math.Floor(expectedBatchCountFloat + 0.5))
+
+	for i := 1; i <= testCount; i += groupSize {
+		cs = append(cs,
+			&Route53Change{
+				Change: route53.Change{
+					Action: aws.String(route53.ChangeActionCreate),
+					ResourceRecordSet: &route53.ResourceRecordSet{
+						Name: aws.String(fmt.Sprintf("host-%d", i)),
+						Type: aws.String("A"),
+						ResourceRecords: []*route53.ResourceRecord{
+							{
+								Value: aws.String("1.2.3.4"),
+							},
+						},
+					},
+				},
+				sizeBytes:  len([]byte("1.2.3.4")),
+				sizeValues: 1,
+			},
+			&Route53Change{
+				Change: route53.Change{
+					Action: aws.String(route53.ChangeActionCreate),
+					ResourceRecordSet: &route53.ResourceRecordSet{
+						Name: aws.String(fmt.Sprintf("host-%d", i)),
+						Type: aws.String("TXT"),
+						ResourceRecords: []*route53.ResourceRecord{
+							{
+								Value: aws.String("txt-record"),
+							},
+						},
+					},
+				},
+				sizeBytes:  len([]byte("txt-record")),
+				sizeValues: 1,
+			},
+		)
+	}
+
+	batchCs := batchChangeSet(cs, defaultBatchChangeSize, defaultBatchChangeSizeBytes, testLimit)
+
+	require.Equal(t, expectedBatchCount, len(batchCs))
+}
+
+func TestAWSBatchChangeSetExceedingValuesLimitUpsert(t *testing.T) {
+	var cs Route53Changes
+	const testCount = 50
+	const testLimit = 100
+	const groupSize = 2
+	// Values for each group multiplied by 2 for Upsert records
+	const testValues = 2 * 2
+	// testCount / groupSize / (testLimit // bytes)
+	var expectedBatchCountFloat = float64(testCount) / float64(groupSize) / float64(testLimit/testValues)
+	// Round up
+	var expectedBatchCount = int(math.Floor(expectedBatchCountFloat + 0.5))
+
+	for i := 1; i <= testCount; i += groupSize {
+		cs = append(cs,
+			&Route53Change{
+				Change: route53.Change{
+					Action: aws.String(route53.ChangeActionUpsert),
+					ResourceRecordSet: &route53.ResourceRecordSet{
+						Name: aws.String(fmt.Sprintf("host-%d", i)),
+						Type: aws.String("A"),
+						ResourceRecords: []*route53.ResourceRecord{
+							{
+								Value: aws.String("1.2.3.4"),
+							},
+						},
+					},
+				},
+				sizeBytes:  len([]byte("1.2.3.4")) * 2,
+				sizeValues: 1,
+			},
+			&Route53Change{
+				Change: route53.Change{
+					Action: aws.String(route53.ChangeActionUpsert),
+					ResourceRecordSet: &route53.ResourceRecordSet{
+						Name: aws.String(fmt.Sprintf("host-%d", i)),
+						Type: aws.String("TXT"),
+						ResourceRecords: []*route53.ResourceRecord{
+							{
+								Value: aws.String("txt-record"),
+							},
+						},
+					},
+				},
+				sizeBytes:  len([]byte("txt-record")) * 2,
+				sizeValues: 1,
+			},
+		)
+	}
+
+	batchCs := batchChangeSet(cs, defaultBatchChangeSize, defaultBatchChangeSizeBytes, testLimit)
+
+	require.Equal(t, expectedBatchCount, len(batchCs))
+}
+
 func validateEndpoints(t *testing.T, provider *AWSProvider, endpoints []*endpoint.Endpoint, expected []*endpoint.Endpoint) {
 	assert.True(t, testutils.SameEndpoints(endpoints, expected), "actual and expected endpoints don't match. %+v:%+v", endpoints, expected)
 
@@ -1935,6 +2154,8 @@ func newAWSProviderWithTagFilter(t *testing.T, domainFilter endpoint.DomainFilte
 	provider := &AWSProvider{
 		client:                client,
 		batchChangeSize:       defaultBatchChangeSize,
+		batchChangeSizeBytes:  defaultBatchChangeSizeBytes,
+		batchChangeSizeValues: defaultBatchChangeSizeValues,
 		batchChangeInterval:   defaultBatchChangeInterval,
 		evaluateTargetHealth:  evaluateTargetHealth,
 		domainFilter:          domainFilter,