Add the ability to configure k8s event rate limit

controller/controller.go

@@ -119,6 +119,8 @@ type Controller struct {
 	nextRunAtMux sync.Mutex
 	// DNS record types that will be considered for management
 	ManagedRecordTypes []string
+	// MinInterval is used as window for batching events
+	MinInterval time.Duration
 }
 
 // RunOnce runs a single iteration of a reconciliation loop.
@@ -165,14 +167,11 @@ func (c *Controller) RunOnce(ctx context.Context) error {
 	return nil
 }
 
-// MinInterval is used as window for batching events
-const MinInterval = 5 * time.Second
-
 // ScheduleRunOnce makes sure execution happens at most once per interval.
 func (c *Controller) ScheduleRunOnce(now time.Time) {
 	c.nextRunAtMux.Lock()
 	defer c.nextRunAtMux.Unlock()
-	c.nextRunAt = now.Add(MinInterval)
+	c.nextRunAt = now.Add(c.MinInterval)
 }
 
 func (c *Controller) ShouldRunOnce(now time.Time) bool {

controller/controller_test.go

@@ -155,7 +155,7 @@ func TestRunOnce(t *testing.T) {
 }
 
 func TestShouldRunOnce(t *testing.T) {
-	ctrl := &Controller{Interval: 10 * time.Minute}
+	ctrl := &Controller{Interval: 10 * time.Minute, MinInterval: 5 * time.Second}
 
 	now := time.Now()
 
@@ -175,7 +175,7 @@ func TestShouldRunOnce(t *testing.T) {
 	assert.False(t, ctrl.ShouldRunOnce(now.Add(100*time.Microsecond)))
 
 	// But after MinInterval we should run reconciliation
-	now = now.Add(MinInterval)
+	now = now.Add(5 * time.Second)
 	assert.True(t, ctrl.ShouldRunOnce(now))
 
 	// But just one time

main.go

@@ -336,6 +336,7 @@ func main() {
 		Interval:           cfg.Interval,
 		DomainFilter:       domainFilter,
 		ManagedRecordTypes: cfg.ManagedDNSRecordTypes,
+		MinInterval:        cfg.MinInterval,
 	}
 
 	if cfg.Once {

pkg/apis/externaldns/types.go

@@ -120,6 +120,7 @@ type Config struct {
 	TXTPrefix                         string
 	TXTSuffix                         string
 	Interval                          time.Duration
+	MinInterval                       time.Duration
 	Once                              bool
 	DryRun                            bool
 	UpdateEvents                      bool
@@ -234,6 +235,7 @@ var defaultConfig = &Config{
 	TXTSuffix:                   "",
 	TXTCacheInterval:            0,
 	TXTWildcardReplacement:      "",
+	MinInterval:                 5 * time.Second,
 	Interval:                    time.Minute,
 	Once:                        false,
 	DryRun:                      false,
@@ -454,6 +456,7 @@ func (cfg *Config) ParseFlags(args []string) error {
 	// Flags related to the main control loop
 	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("min-interval", "The minimum interval between two consecutive synchronizations triggered from kubernetes events in duration format (default: 5s)").Default(defaultConfig.MinInterval.String()).DurationVar(&cfg.MinInterval)
 	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)
 	app.Flag("events", "When enabled, in addition to running every interval, the reconciliation loop will get triggered when supported sources change (default: disabled)").BoolVar(&cfg.UpdateEvents)

pkg/apis/externaldns/types_test.go

@@ -90,6 +90,7 @@ var (
 		TXTPrefix:                   "",
 		TXTCacheInterval:            0,
 		Interval:                    time.Minute,
+		MinInterval:                 5 * time.Second,
 		Once:                        false,
 		DryRun:                      false,
 		UpdateEvents:                false,
@@ -149,8 +150,8 @@ var (
 		AkamaiClientToken:           "o184671d5307a388180fbf7f11dbdf46",
 		AkamaiClientSecret:          "o184671d5307a388180fbf7f11dbdf46",
 		AkamaiAccessToken:           "o184671d5307a388180fbf7f11dbdf46",
-	        AkamaiEdgercPath:            "/home/test/.edgerc",
-        	AkamaiEdgercSection:         "default",
+		AkamaiEdgercPath:            "/home/test/.edgerc",
+		AkamaiEdgercSection:         "default",
 		InfobloxGridHost:            "127.0.0.1",
 		InfobloxWapiPort:            8443,
 		InfobloxWapiUsername:        "infoblox",
@@ -175,6 +176,7 @@ var (
 		TXTPrefix:                   "associated-txt-record",
 		TXTCacheInterval:            12 * time.Hour,
 		Interval:                    10 * time.Minute,
+		MinInterval:                 50 * time.Second,
 		Once:                        true,
 		DryRun:                      true,
 		UpdateEvents:                true,
@@ -287,6 +289,7 @@ func TestParseFlags(t *testing.T) {
 				"--txt-prefix=associated-txt-record",
 				"--txt-cache-interval=12h",
 				"--interval=10m",
+				"--min-interval=50s",
 				"--once",
 				"--dry-run",
 				"--events",
@@ -378,6 +381,7 @@ func TestParseFlags(t *testing.T) {
 				"EXTERNAL_DNS_TXT_PREFIX":                      "associated-txt-record",
 				"EXTERNAL_DNS_TXT_CACHE_INTERVAL":              "12h",
 				"EXTERNAL_DNS_INTERVAL":                        "10m",
+				"EXTERNAL_DNS_MIN_INTERVAL":                    "50s",
 				"EXTERNAL_DNS_ONCE":                            "1",
 				"EXTERNAL_DNS_DRY_RUN":                         "1",
 				"EXTERNAL_DNS_EVENTS":                          "1",