Add the ability to configure k8s event rate limit

Currently, the minimum delay between 2 kubernetes events handling is
hard-coded to 5s.

This may cause higher synchronization rates and higher DNS provider API
calls when handling an important number of kubernetes events at once.

Give the opportunity to configure this delay so service owners can
define the acceptable thresholds on their side

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",