Merge pull request #2013 from tjamet/feat/flag-min-sync

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
+	// MinEventSyncInterval is used as window for batching events
+	MinEventSyncInterval 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.MinEventSyncInterval)
 }
 
 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, MinEventSyncInterval: 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

@@ -334,12 +334,13 @@ func main() {
 	}
 
 	ctrl := controller.Controller{
-		Source:             endpointsSource,
-		Registry:           r,
-		Policy:             policy,
-		Interval:           cfg.Interval,
-		DomainFilter:       domainFilter,
-		ManagedRecordTypes: cfg.ManagedDNSRecordTypes,
+		Source:               endpointsSource,
+		Registry:             r,
+		Policy:               policy,
+		Interval:             cfg.Interval,
+		DomainFilter:         domainFilter,
+		ManagedRecordTypes:   cfg.ManagedDNSRecordTypes,
+		MinEventSyncInterval: cfg.MinEventSyncInterval,
 	}
 
 	if cfg.Once {

pkg/apis/externaldns/types.go

@@ -122,6 +122,7 @@ type Config struct {
 	TXTPrefix                         string
 	TXTSuffix                         string
 	Interval                          time.Duration
+	MinEventSyncInterval              time.Duration
 	Once                              bool
 	DryRun                            bool
 	UpdateEvents                      bool
@@ -239,6 +240,7 @@ var defaultConfig = &Config{
 	TXTSuffix:                   "",
 	TXTCacheInterval:            0,
 	TXTWildcardReplacement:      "",
+	MinEventSyncInterval:        5 * time.Second,
 	Interval:                    time.Minute,
 	Once:                        false,
 	DryRun:                      false,
@@ -464,6 +466,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-event-sync-interval", "The minimum interval between two consecutive synchronizations triggered from kubernetes events in duration format (default: 5s)").Default(defaultConfig.MinEventSyncInterval.String()).DurationVar(&cfg.MinEventSyncInterval)
 	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

@@ -92,6 +92,7 @@ var (
 		TXTPrefix:                   "",
 		TXTCacheInterval:            0,
 		Interval:                    time.Minute,
+		MinEventSyncInterval:        5 * time.Second,
 		Once:                        false,
 		DryRun:                      false,
 		UpdateEvents:                false,
@@ -153,8 +154,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",
@@ -179,6 +180,7 @@ var (
 		TXTPrefix:                   "associated-txt-record",
 		TXTCacheInterval:            12 * time.Hour,
 		Interval:                    10 * time.Minute,
+		MinEventSyncInterval:        50 * time.Second,
 		Once:                        true,
 		DryRun:                      true,
 		UpdateEvents:                true,
@@ -293,6 +295,7 @@ func TestParseFlags(t *testing.T) {
 				"--txt-prefix=associated-txt-record",
 				"--txt-cache-interval=12h",
 				"--interval=10m",
+				"--min-event-sync-interval=50s",
 				"--once",
 				"--dry-run",
 				"--events",
@@ -386,6 +389,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_EVENT_SYNC_INTERVAL":         "50s",
 				"EXTERNAL_DNS_ONCE":                            "1",
 				"EXTERNAL_DNS_DRY_RUN":                         "1",
 				"EXTERNAL_DNS_EVENTS":                          "1",