From 6da9d363f36d512bdb6f1c022be2da07d47226a6 Mon Sep 17 00:00:00 2001 From: Antonio Ojea Date: Mon, 28 Apr 2025 20:47:48 +0200 Subject: [PATCH 1/5] Copy BoundedFrequencyRunner to kube-proxy --- pkg/proxy/iptables/proxier.go | 6 +- pkg/proxy/iptables/proxier_test.go | 4 +- pkg/proxy/ipvs/proxier.go | 6 +- pkg/proxy/ipvs/proxier_test.go | 4 +- pkg/proxy/nftables/proxier.go | 6 +- pkg/proxy/nftables/proxier_test.go | 4 +- pkg/proxy/runner/bounded_frequency_runner.go | 313 ++++++++++++ .../runner/bounded_frequency_runner_test.go | 453 ++++++++++++++++++ pkg/proxy/winkernel/proxier.go | 6 +- 9 files changed, 784 insertions(+), 18 deletions(-) create mode 100644 pkg/proxy/runner/bounded_frequency_runner.go create mode 100644 pkg/proxy/runner/bounded_frequency_runner_test.go diff --git a/pkg/proxy/iptables/proxier.go b/pkg/proxy/iptables/proxier.go index d50e6c06bd2..600ab9b62bf 100644 --- a/pkg/proxy/iptables/proxier.go +++ b/pkg/proxy/iptables/proxier.go @@ -46,9 +46,9 @@ import ( "k8s.io/kubernetes/pkg/proxy/healthcheck" "k8s.io/kubernetes/pkg/proxy/metaproxier" "k8s.io/kubernetes/pkg/proxy/metrics" + "k8s.io/kubernetes/pkg/proxy/runner" proxyutil "k8s.io/kubernetes/pkg/proxy/util" "k8s.io/kubernetes/pkg/proxy/util/nfacct" - "k8s.io/kubernetes/pkg/util/async" utiliptables "k8s.io/kubernetes/pkg/util/iptables" ) @@ -155,7 +155,7 @@ type Proxier struct { lastFullSync time.Time needFullSync bool initialized int32 - syncRunner *async.BoundedFrequencyRunner // governs calls to syncProxyRules + syncRunner *runner.BoundedFrequencyRunner // governs calls to syncProxyRules syncPeriod time.Duration lastIPTablesCleanup time.Time @@ -312,7 +312,7 @@ func NewProxier(ctx context.Context, // We pass syncPeriod to ipt.Monitor, which will call us only if it needs to. // We need to pass *some* maxInterval to NewBoundedFrequencyRunner anyway though. // time.Hour is arbitrary. - proxier.syncRunner = async.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, proxyutil.FullSyncPeriod, burstSyncs) + proxier.syncRunner = runner.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, proxyutil.FullSyncPeriod, burstSyncs) go ipt.Monitor(kubeProxyCanaryChain, []utiliptables.Table{utiliptables.TableMangle, utiliptables.TableNAT, utiliptables.TableFilter}, proxier.forceSyncProxyRules, syncPeriod, wait.NeverStop) diff --git a/pkg/proxy/iptables/proxier_test.go b/pkg/proxy/iptables/proxier_test.go index ef1747396fc..430d7c34a77 100644 --- a/pkg/proxy/iptables/proxier_test.go +++ b/pkg/proxy/iptables/proxier_test.go @@ -55,9 +55,9 @@ import ( "k8s.io/kubernetes/pkg/proxy/conntrack" "k8s.io/kubernetes/pkg/proxy/healthcheck" "k8s.io/kubernetes/pkg/proxy/metrics" + "k8s.io/kubernetes/pkg/proxy/runner" proxyutil "k8s.io/kubernetes/pkg/proxy/util" proxyutiltest "k8s.io/kubernetes/pkg/proxy/util/testing" - "k8s.io/kubernetes/pkg/util/async" utiliptables "k8s.io/kubernetes/pkg/util/iptables" iptablestest "k8s.io/kubernetes/pkg/util/iptables/testing" netutils "k8s.io/utils/net" @@ -144,7 +144,7 @@ func NewFakeProxier(ipt utiliptables.Interface) *Proxier { }, } p.setInitialized(true) - p.syncRunner = async.NewBoundedFrequencyRunner("test-sync-runner", p.syncProxyRules, 0, time.Minute, 1) + p.syncRunner = runner.NewBoundedFrequencyRunner("test-sync-runner", p.syncProxyRules, 0, time.Minute, 1) return p } diff --git a/pkg/proxy/ipvs/proxier.go b/pkg/proxy/ipvs/proxier.go index 51a6cf3faa5..7995730d320 100644 --- a/pkg/proxy/ipvs/proxier.go +++ b/pkg/proxy/ipvs/proxier.go @@ -49,8 +49,8 @@ import ( utilipvs "k8s.io/kubernetes/pkg/proxy/ipvs/util" "k8s.io/kubernetes/pkg/proxy/metaproxier" "k8s.io/kubernetes/pkg/proxy/metrics" + "k8s.io/kubernetes/pkg/proxy/runner" proxyutil "k8s.io/kubernetes/pkg/proxy/util" - "k8s.io/kubernetes/pkg/util/async" utiliptables "k8s.io/kubernetes/pkg/util/iptables" utilkernel "k8s.io/kubernetes/pkg/util/kernel" netutils "k8s.io/utils/net" @@ -187,7 +187,7 @@ type Proxier struct { endpointSlicesSynced bool servicesSynced bool initialized int32 - syncRunner *async.BoundedFrequencyRunner // governs calls to syncProxyRules + syncRunner *runner.BoundedFrequencyRunner // governs calls to syncProxyRules // These are effectively const and do not need the mutex to be held. syncPeriod time.Duration @@ -402,7 +402,7 @@ func NewProxier( } burstSyncs := 2 logger.V(2).Info("ipvs sync params", "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod, "burstSyncs", burstSyncs) - proxier.syncRunner = async.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, burstSyncs) + proxier.syncRunner = runner.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, burstSyncs) proxier.gracefuldeleteManager.Run() return proxier, nil } diff --git a/pkg/proxy/ipvs/proxier_test.go b/pkg/proxy/ipvs/proxier_test.go index 09e95316d9b..5befa4d2476 100644 --- a/pkg/proxy/ipvs/proxier_test.go +++ b/pkg/proxy/ipvs/proxier_test.go @@ -53,9 +53,9 @@ import ( utilipvs "k8s.io/kubernetes/pkg/proxy/ipvs/util" ipvstest "k8s.io/kubernetes/pkg/proxy/ipvs/util/testing" "k8s.io/kubernetes/pkg/proxy/metrics" + "k8s.io/kubernetes/pkg/proxy/runner" proxyutil "k8s.io/kubernetes/pkg/proxy/util" proxyutiltest "k8s.io/kubernetes/pkg/proxy/util/testing" - "k8s.io/kubernetes/pkg/util/async" utiliptables "k8s.io/kubernetes/pkg/util/iptables" iptablestest "k8s.io/kubernetes/pkg/util/iptables/testing" "k8s.io/kubernetes/test/utils/ktesting" @@ -167,7 +167,7 @@ func NewFakeProxier(ctx context.Context, ipt utiliptables.Interface, ipvs utilip ipFamily: ipFamily, } p.setInitialized(true) - p.syncRunner = async.NewBoundedFrequencyRunner("test-sync-runner", p.syncProxyRules, 0, time.Minute, 1) + p.syncRunner = runner.NewBoundedFrequencyRunner("test-sync-runner", p.syncProxyRules, 0, time.Minute, 1) return p } diff --git a/pkg/proxy/nftables/proxier.go b/pkg/proxy/nftables/proxier.go index 052d4fba8c2..6e2ce149fb3 100644 --- a/pkg/proxy/nftables/proxier.go +++ b/pkg/proxy/nftables/proxier.go @@ -49,8 +49,8 @@ import ( "k8s.io/kubernetes/pkg/proxy/healthcheck" "k8s.io/kubernetes/pkg/proxy/metaproxier" "k8s.io/kubernetes/pkg/proxy/metrics" + "k8s.io/kubernetes/pkg/proxy/runner" proxyutil "k8s.io/kubernetes/pkg/proxy/util" - "k8s.io/kubernetes/pkg/util/async" utilkernel "k8s.io/kubernetes/pkg/util/kernel" netutils "k8s.io/utils/net" "k8s.io/utils/ptr" @@ -164,7 +164,7 @@ type Proxier struct { lastFullSync time.Time needFullSync bool initialized int32 - syncRunner *async.BoundedFrequencyRunner // governs calls to syncProxyRules + syncRunner *runner.BoundedFrequencyRunner // governs calls to syncProxyRules syncPeriod time.Duration flushed bool @@ -276,7 +276,7 @@ func NewProxier(ctx context.Context, burstSyncs := 2 logger.V(2).Info("NFTables sync params", "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod, "burstSyncs", burstSyncs) // We need to pass *some* maxInterval to NewBoundedFrequencyRunner. time.Hour is arbitrary. - proxier.syncRunner = async.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, proxyutil.FullSyncPeriod, burstSyncs) + proxier.syncRunner = runner.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, proxyutil.FullSyncPeriod, burstSyncs) return proxier, nil } diff --git a/pkg/proxy/nftables/proxier_test.go b/pkg/proxy/nftables/proxier_test.go index 52207148544..cc19185639e 100644 --- a/pkg/proxy/nftables/proxier_test.go +++ b/pkg/proxy/nftables/proxier_test.go @@ -46,9 +46,9 @@ import ( "k8s.io/kubernetes/pkg/proxy/conntrack" "k8s.io/kubernetes/pkg/proxy/healthcheck" "k8s.io/kubernetes/pkg/proxy/metrics" + "k8s.io/kubernetes/pkg/proxy/runner" proxyutil "k8s.io/kubernetes/pkg/proxy/util" proxyutiltest "k8s.io/kubernetes/pkg/proxy/util/testing" - "k8s.io/kubernetes/pkg/util/async" netutils "k8s.io/utils/net" "k8s.io/utils/ptr" "sigs.k8s.io/knftables" @@ -142,7 +142,7 @@ func NewFakeProxier(ipFamily v1.IPFamily) (*knftables.Fake, *Proxier) { serviceNodePorts: newNFTElementStorage("map", serviceNodePortsMap), } p.setInitialized(true) - p.syncRunner = async.NewBoundedFrequencyRunner("test-sync-runner", p.syncProxyRules, 0, time.Minute, 1) + p.syncRunner = runner.NewBoundedFrequencyRunner("test-sync-runner", p.syncProxyRules, 0, time.Minute, 1) return nft, p } diff --git a/pkg/proxy/runner/bounded_frequency_runner.go b/pkg/proxy/runner/bounded_frequency_runner.go new file mode 100644 index 00000000000..8b13c2fd4b5 --- /dev/null +++ b/pkg/proxy/runner/bounded_frequency_runner.go @@ -0,0 +1,313 @@ +/* +Copyright 2017 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package runner + +import ( + "fmt" + "sync" + "time" + + "k8s.io/client-go/util/flowcontrol" + + "k8s.io/klog/v2" +) + +// BoundedFrequencyRunner manages runs of a user-provided function. +// See NewBoundedFrequencyRunner for examples. +type BoundedFrequencyRunner struct { + name string // the name of this instance + minInterval time.Duration // the min time between runs, modulo bursts + maxInterval time.Duration // the max time between runs + + run chan struct{} // try an async run + + mu sync.Mutex // guards runs of fn and all mutations + fn func() // function to run + lastRun time.Time // time of last run + timer timer // timer for deferred runs + limiter rateLimiter // rate limiter for on-demand runs + + retry chan struct{} // schedule a retry + retryMu sync.Mutex // guards retryTime + retryTime time.Time // when to retry +} + +// designed so that flowcontrol.RateLimiter satisfies +type rateLimiter interface { + TryAccept() bool + Stop() +} + +type nullLimiter struct{} + +func (nullLimiter) TryAccept() bool { + return true +} + +func (nullLimiter) Stop() {} + +var _ rateLimiter = nullLimiter{} + +// for testing +type timer interface { + // C returns the timer's selectable channel. + C() <-chan time.Time + + // See time.Timer.Reset. + Reset(d time.Duration) bool + + // See time.Timer.Stop. + Stop() bool + + // See time.Now. + Now() time.Time + + // Remaining returns the time until the timer will go off (if it is running). + Remaining() time.Duration + + // See time.Since. + Since(t time.Time) time.Duration + + // See time.Sleep. + Sleep(d time.Duration) +} + +// implement our timer in terms of std time.Timer. +type realTimer struct { + timer *time.Timer + next time.Time +} + +func (rt *realTimer) C() <-chan time.Time { + return rt.timer.C +} + +func (rt *realTimer) Reset(d time.Duration) bool { + rt.next = time.Now().Add(d) + return rt.timer.Reset(d) +} + +func (rt *realTimer) Stop() bool { + return rt.timer.Stop() +} + +func (rt *realTimer) Now() time.Time { + return time.Now() +} + +func (rt *realTimer) Remaining() time.Duration { + return rt.next.Sub(time.Now()) +} + +func (rt *realTimer) Since(t time.Time) time.Duration { + return time.Since(t) +} + +func (rt *realTimer) Sleep(d time.Duration) { + time.Sleep(d) +} + +var _ timer = &realTimer{} + +// NewBoundedFrequencyRunner creates a new BoundedFrequencyRunner instance, +// which will manage runs of the specified function. +// +// All runs will be async to the caller of BoundedFrequencyRunner.Run, but +// multiple runs are serialized. If the function needs to hold locks, it must +// take them internally. +// +// Runs of the function will have at least minInterval between them (from +// completion to next start), except that up to bursts may be allowed. Burst +// runs are "accumulated" over time, one per minInterval up to burstRuns total. +// This can be used, for example, to mitigate the impact of expensive operations +// being called in response to user-initiated operations. Run requests that +// would violate the minInterval are coalesced and run at the next opportunity. +// +// The function will be run at least once per maxInterval. For example, this can +// force periodic refreshes of state in the absence of anyone calling Run. +// +// Examples: +// +// NewBoundedFrequencyRunner("name", fn, time.Second, 5*time.Second, 1) +// - fn will have at least 1 second between runs +// - fn will have no more than 5 seconds between runs +// +// NewBoundedFrequencyRunner("name", fn, 3*time.Second, 10*time.Second, 3) +// - fn will have at least 3 seconds between runs, with up to 3 burst runs +// - fn will have no more than 10 seconds between runs +// +// The maxInterval must be greater than or equal to the minInterval, If the +// caller passes a maxInterval less than minInterval, this function will panic. +func NewBoundedFrequencyRunner(name string, fn func(), minInterval, maxInterval time.Duration, burstRuns int) *BoundedFrequencyRunner { + timer := &realTimer{timer: time.NewTimer(0)} // will tick immediately + <-timer.C() // consume the first tick + return construct(name, fn, minInterval, maxInterval, burstRuns, timer) +} + +// Make an instance with dependencies injected. +func construct(name string, fn func(), minInterval, maxInterval time.Duration, burstRuns int, timer timer) *BoundedFrequencyRunner { + if maxInterval < minInterval { + panic(fmt.Sprintf("%s: maxInterval (%v) must be >= minInterval (%v)", name, maxInterval, minInterval)) + } + if timer == nil { + panic(fmt.Sprintf("%s: timer must be non-nil", name)) + } + + bfr := &BoundedFrequencyRunner{ + name: name, + fn: fn, + minInterval: minInterval, + maxInterval: maxInterval, + run: make(chan struct{}, 1), + retry: make(chan struct{}, 1), + timer: timer, + } + if minInterval == 0 { + bfr.limiter = nullLimiter{} + } else { + // allow burst updates in short succession + qps := float32(time.Second) / float32(minInterval) + bfr.limiter = flowcontrol.NewTokenBucketRateLimiterWithClock(qps, burstRuns, timer) + } + return bfr +} + +// Loop handles the periodic timer and run requests. This is expected to be +// called as a goroutine. +func (bfr *BoundedFrequencyRunner) Loop(stop <-chan struct{}) { + klog.V(3).Infof("%s Loop running", bfr.name) + bfr.timer.Reset(bfr.maxInterval) + for { + select { + case <-stop: + bfr.stop() + klog.V(3).Infof("%s Loop stopping", bfr.name) + return + case <-bfr.timer.C(): + bfr.tryRun() + case <-bfr.run: + bfr.tryRun() + case <-bfr.retry: + bfr.doRetry() + } + } +} + +// Run the function as soon as possible. If this is called while Loop is not +// running, the call may be deferred indefinitely. +// If there is already a queued request to call the underlying function, it +// may be dropped - it is just guaranteed that we will try calling the +// underlying function as soon as possible starting from now. +func (bfr *BoundedFrequencyRunner) Run() { + // If it takes a lot of time to run the underlying function, noone is really + // processing elements from channel. So to avoid blocking here on the + // putting element to it, we simply skip it if there is already an element + // in it. + select { + case bfr.run <- struct{}{}: + default: + } +} + +// RetryAfter ensures that the function will run again after no later than interval. This +// can be called from inside a run of the BoundedFrequencyRunner's function, or +// asynchronously. +func (bfr *BoundedFrequencyRunner) RetryAfter(interval time.Duration) { + // This could be called either with or without bfr.mu held, so we can't grab that + // lock, and therefore we can't update the timer directly. + + // If the Loop thread is currently running fn then it may be a while before it + // processes our retry request. But we want to retry at interval from now, not at + // interval from "whenever doRetry eventually gets called". So we convert to + // absolute time. + retryTime := bfr.timer.Now().Add(interval) + + // We can't just write retryTime to a channel because there could be multiple + // RetryAfter calls before Loop gets a chance to read from the channel. So we + // record the soonest requested retry time in bfr.retryTime and then only signal + // the Loop thread once, just like Run does. + bfr.retryMu.Lock() + defer bfr.retryMu.Unlock() + if !bfr.retryTime.IsZero() && bfr.retryTime.Before(retryTime) { + return + } + bfr.retryTime = retryTime + + select { + case bfr.retry <- struct{}{}: + default: + } +} + +// assumes the lock is not held +func (bfr *BoundedFrequencyRunner) stop() { + bfr.mu.Lock() + defer bfr.mu.Unlock() + bfr.limiter.Stop() + bfr.timer.Stop() +} + +// assumes the lock is not held +func (bfr *BoundedFrequencyRunner) doRetry() { + bfr.mu.Lock() + defer bfr.mu.Unlock() + bfr.retryMu.Lock() + defer bfr.retryMu.Unlock() + + if bfr.retryTime.IsZero() { + return + } + + // Timer wants an interval not an absolute time, so convert retryTime back now + retryInterval := bfr.retryTime.Sub(bfr.timer.Now()) + bfr.retryTime = time.Time{} + if retryInterval < bfr.timer.Remaining() { + klog.V(3).Infof("%s: retrying in %v", bfr.name, retryInterval) + bfr.timer.Stop() + bfr.timer.Reset(retryInterval) + } +} + +// assumes the lock is not held +func (bfr *BoundedFrequencyRunner) tryRun() { + bfr.mu.Lock() + defer bfr.mu.Unlock() + + if bfr.limiter.TryAccept() { + // We're allowed to run the function right now. + bfr.fn() + bfr.lastRun = bfr.timer.Now() + bfr.timer.Stop() + bfr.timer.Reset(bfr.maxInterval) + klog.V(3).Infof("%s: ran, next possible in %v, periodic in %v", bfr.name, bfr.minInterval, bfr.maxInterval) + return + } + + // It can't run right now, figure out when it can run next. + elapsed := bfr.timer.Since(bfr.lastRun) // how long since last run + nextPossible := bfr.minInterval - elapsed // time to next possible run + nextScheduled := bfr.timer.Remaining() // time to next scheduled run + klog.V(4).Infof("%s: %v since last run, possible in %v, scheduled in %v", bfr.name, elapsed, nextPossible, nextScheduled) + + // It's hard to avoid race conditions in the unit tests unless we always reset + // the timer here, even when it's unchanged + if nextPossible < nextScheduled { + nextScheduled = nextPossible + } + bfr.timer.Stop() + bfr.timer.Reset(nextScheduled) +} diff --git a/pkg/proxy/runner/bounded_frequency_runner_test.go b/pkg/proxy/runner/bounded_frequency_runner_test.go new file mode 100644 index 00000000000..e4f4e4933d2 --- /dev/null +++ b/pkg/proxy/runner/bounded_frequency_runner_test.go @@ -0,0 +1,453 @@ +/* +Copyright 2017 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package runner + +import ( + "sync" + "testing" + "time" +) + +// Track calls to the managed function. +type receiver struct { + lock sync.Mutex + run bool + retryFn func() +} + +func (r *receiver) F() { + r.lock.Lock() + defer r.lock.Unlock() + r.run = true + + if r.retryFn != nil { + r.retryFn() + r.retryFn = nil + } +} + +func (r *receiver) reset() bool { + r.lock.Lock() + defer r.lock.Unlock() + was := r.run + r.run = false + return was +} + +func (r *receiver) setRetryFn(retryFn func()) { + r.lock.Lock() + defer r.lock.Unlock() + r.retryFn = retryFn +} + +// A single change event in the fake timer. +type timerUpdate struct { + active bool + next time.Duration // iff active == true +} + +// Fake time. +type fakeTimer struct { + c chan time.Time + + lock sync.Mutex + now time.Time + timeout time.Time + active bool + + updated chan timerUpdate +} + +func newFakeTimer() *fakeTimer { + ft := &fakeTimer{ + now: time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC), + c: make(chan time.Time), + updated: make(chan timerUpdate), + } + return ft +} + +func (ft *fakeTimer) C() <-chan time.Time { + return ft.c +} + +func (ft *fakeTimer) Reset(in time.Duration) bool { + ft.lock.Lock() + defer ft.lock.Unlock() + + was := ft.active + ft.active = true + ft.timeout = ft.now.Add(in) + ft.updated <- timerUpdate{ + active: true, + next: in, + } + return was +} + +func (ft *fakeTimer) Stop() bool { + ft.lock.Lock() + defer ft.lock.Unlock() + + was := ft.active + ft.active = false + ft.updated <- timerUpdate{ + active: false, + } + return was +} + +func (ft *fakeTimer) Now() time.Time { + ft.lock.Lock() + defer ft.lock.Unlock() + + return ft.now +} + +func (ft *fakeTimer) Remaining() time.Duration { + ft.lock.Lock() + defer ft.lock.Unlock() + + return ft.timeout.Sub(ft.now) +} + +func (ft *fakeTimer) Since(t time.Time) time.Duration { + ft.lock.Lock() + defer ft.lock.Unlock() + + return ft.now.Sub(t) +} + +func (ft *fakeTimer) Sleep(d time.Duration) { + // ft.advance grabs ft.lock + ft.advance(d) +} + +// advance the current time. +func (ft *fakeTimer) advance(d time.Duration) { + ft.lock.Lock() + defer ft.lock.Unlock() + + ft.now = ft.now.Add(d) + if ft.active && !ft.now.Before(ft.timeout) { + ft.active = false + ft.c <- ft.timeout + } +} + +// return the calling line number (for printing) +// test the timer's state +func checkTimer(name string, t *testing.T, upd timerUpdate, active bool, next time.Duration) { + if upd.active != active { + t.Fatalf("%s: expected timer active=%v", name, active) + } + if active && upd.next != next { + t.Fatalf("%s: expected timer to be %v, got %v", name, next, upd.next) + } +} + +// test and reset the receiver's state +func checkReceiver(name string, t *testing.T, receiver *receiver, expected bool) { + triggered := receiver.reset() + if expected && !triggered { + t.Fatalf("%s: function should have been called", name) + } else if !expected && triggered { + t.Fatalf("%s: function should not have been called", name) + } +} + +// Durations embedded in test cases depend on these. +var minInterval = 1 * time.Second +var maxInterval = 10 * time.Second + +func waitForReset(name string, t *testing.T, timer *fakeTimer, obj *receiver, expectCall bool, expectNext time.Duration) { + upd := <-timer.updated // wait for stop + checkReceiver(name, t, obj, expectCall) + checkReceiver(name, t, obj, false) // prove post-condition + checkTimer(name, t, upd, false, 0) + upd = <-timer.updated // wait for reset + checkTimer(name, t, upd, true, expectNext) +} + +func waitForRun(name string, t *testing.T, timer *fakeTimer, obj *receiver) { + waitForReset(name, t, timer, obj, true, maxInterval) +} + +func waitForRunWithRetry(name string, t *testing.T, timer *fakeTimer, obj *receiver, expectNext time.Duration) { + // It will first get reset as with a normal run, and then get set again + waitForRun(name, t, timer, obj) + waitForReset(name, t, timer, obj, false, expectNext) +} + +func waitForDefer(name string, t *testing.T, timer *fakeTimer, obj *receiver, expectNext time.Duration) { + waitForReset(name, t, timer, obj, false, expectNext) +} + +func waitForNothing(name string, t *testing.T, timer *fakeTimer, obj *receiver) { + select { + case <-timer.c: + t.Fatalf("%s: unexpected timer tick", name) + case upd := <-timer.updated: + t.Fatalf("%s: unexpected timer update %v", name, upd) + default: + } + checkReceiver(name, t, obj, false) +} + +func Test_BoundedFrequencyRunnerNoBurst(t *testing.T) { + obj := &receiver{} + timer := newFakeTimer() + runner := construct("test-runner", obj.F, minInterval, maxInterval, 1, timer) + stop := make(chan struct{}) + + var upd timerUpdate + + // Start. + go runner.Loop(stop) + upd = <-timer.updated // wait for initial time to be set to max + checkTimer("init", t, upd, true, maxInterval) + checkReceiver("init", t, obj, false) + + // Run once, immediately. + // rel=0ms + runner.Run() + waitForRun("first run", t, timer, obj) + + // Run again, before minInterval expires. + timer.advance(500 * time.Millisecond) // rel=500ms + runner.Run() + waitForDefer("too soon after first", t, timer, obj, 500*time.Millisecond) + + // Run again, before minInterval expires. + timer.advance(499 * time.Millisecond) // rel=999ms + runner.Run() + waitForDefer("still too soon after first", t, timer, obj, 1*time.Millisecond) + + // Do the deferred run + timer.advance(1 * time.Millisecond) // rel=1000ms + waitForRun("second run", t, timer, obj) + + // Try again immediately + runner.Run() + waitForDefer("too soon after second", t, timer, obj, 1*time.Second) + + // Run again, before minInterval expires. + timer.advance(1 * time.Millisecond) // rel=1ms + runner.Run() + waitForDefer("still too soon after second", t, timer, obj, 999*time.Millisecond) + + // Ensure that we don't run again early + timer.advance(998 * time.Millisecond) // rel=999ms + waitForNothing("premature", t, timer, obj) + + // Do the deferred run + timer.advance(1 * time.Millisecond) // rel=1000ms + waitForRun("third run", t, timer, obj) + + // Let minInterval pass, but there are no runs queued + timer.advance(1 * time.Second) // rel=1000ms + waitForNothing("minInterval", t, timer, obj) + + // Let maxInterval pass + timer.advance(9 * time.Second) // rel=10000ms + waitForRun("maxInterval", t, timer, obj) + + // Run again, before minInterval expires. + timer.advance(1 * time.Millisecond) // rel=1ms + runner.Run() + waitForDefer("too soon after maxInterval run", t, timer, obj, 999*time.Millisecond) + + // Let minInterval pass + timer.advance(999 * time.Millisecond) // rel=1000ms + waitForRun("fifth run", t, timer, obj) + + // Clean up. + stop <- struct{}{} + // a message is sent to time.updated in func Stop() at the end of the child goroutine + // to terminate the child, a receive on time.updated is needed here + <-timer.updated +} + +func Test_BoundedFrequencyRunnerBurst(t *testing.T) { + obj := &receiver{} + timer := newFakeTimer() + runner := construct("test-runner", obj.F, minInterval, maxInterval, 2, timer) + stop := make(chan struct{}) + + var upd timerUpdate + + // Start. + go runner.Loop(stop) + upd = <-timer.updated // wait for initial time to be set to max + checkTimer("init", t, upd, true, maxInterval) + checkReceiver("init", t, obj, false) + + // Run once, immediately. + // abs=0ms, rel=0ms + runner.Run() + waitForRun("first run", t, timer, obj) + + // Run again, before minInterval expires, with burst. + timer.advance(1 * time.Millisecond) // abs=1ms, rel=1ms + runner.Run() + waitForRun("second run", t, timer, obj) + + // Run again, before minInterval expires. + timer.advance(498 * time.Millisecond) // abs=499ms, rel=498ms + runner.Run() + waitForDefer("too soon after second", t, timer, obj, 502*time.Millisecond) + + // Run again, before minInterval expires. + timer.advance(1 * time.Millisecond) // abs=500ms, rel=499ms + runner.Run() + waitForDefer("too soon after second 2", t, timer, obj, 501*time.Millisecond) + + // Run again, before minInterval expires. + timer.advance(1 * time.Millisecond) // abs=501ms, rel=500ms + runner.Run() + waitForDefer("too soon after second 3", t, timer, obj, 500*time.Millisecond) + + // Advance timer enough to replenish bursts, but not enough to be minInterval + // after the last run + timer.advance(499 * time.Millisecond) // abs=1000ms, rel=999ms + waitForNothing("not minInterval", t, timer, obj) + runner.Run() + waitForRun("third run", t, timer, obj) + + // Run again, before minInterval expires. + timer.advance(1 * time.Millisecond) // abs=1001ms, rel=1ms + runner.Run() + waitForDefer("too soon after third", t, timer, obj, 999*time.Millisecond) + + // Run again, before minInterval expires. + timer.advance(998 * time.Millisecond) // abs=1999ms, rel=999ms + runner.Run() + waitForDefer("too soon after third 2", t, timer, obj, 1*time.Millisecond) + + // Advance and do the deferred run + timer.advance(1 * time.Millisecond) // abs=2000ms, rel=1000ms + waitForRun("fourth run", t, timer, obj) + + // Run again, once burst has fully replenished. + timer.advance(2 * time.Second) // abs=4000ms, rel=2000ms + runner.Run() + waitForRun("fifth run", t, timer, obj) + runner.Run() + waitForRun("sixth run", t, timer, obj) + runner.Run() + waitForDefer("too soon after sixth", t, timer, obj, 1*time.Second) + + // Wait until minInterval after the last run + timer.advance(1 * time.Second) // abs=5000ms, rel=1000ms + waitForRun("seventh run", t, timer, obj) + + // Wait for maxInterval + timer.advance(10 * time.Second) // abs=15000ms, rel=10000ms + waitForRun("maxInterval", t, timer, obj) + + // Clean up. + stop <- struct{}{} + // a message is sent to time.updated in func Stop() at the end of the child goroutine + // to terminate the child, a receive on time.updated is needed here + <-timer.updated +} + +func Test_BoundedFrequencyRunnerRetryAfter(t *testing.T) { + obj := &receiver{} + timer := newFakeTimer() + runner := construct("test-runner", obj.F, minInterval, maxInterval, 1, timer) + stop := make(chan struct{}) + + var upd timerUpdate + + // Start. + go runner.Loop(stop) + upd = <-timer.updated // wait for initial time to be set to max + checkTimer("init", t, upd, true, maxInterval) + checkReceiver("init", t, obj, false) + + // Run once, immediately, and queue a retry + // rel=0ms + obj.setRetryFn(func() { runner.RetryAfter(5 * time.Second) }) + runner.Run() + waitForRunWithRetry("first run", t, timer, obj, 5*time.Second) + + // Nothing happens... + timer.advance(time.Second) // rel=1000ms + waitForNothing("minInterval, nothing queued", t, timer, obj) + + // After retryInterval, function is called + timer.advance(4 * time.Second) // rel=5000ms + waitForRun("retry", t, timer, obj) + + // Run again, before minInterval expires. + timer.advance(499 * time.Millisecond) // rel=499ms + runner.Run() + waitForDefer("too soon after retry", t, timer, obj, 501*time.Millisecond) + + // Do the deferred run, queue another retry after it returns + timer.advance(501 * time.Millisecond) // rel=1000ms + runner.RetryAfter(5 * time.Second) + waitForRunWithRetry("second run", t, timer, obj, 5*time.Second) + + // Wait for minInterval to pass + timer.advance(time.Second) // rel=1000ms + waitForNothing("minInterval, nothing queued", t, timer, obj) + + // Now do another run + runner.Run() + waitForRun("third run", t, timer, obj) + + // Retry was cancelled because we already ran + timer.advance(4 * time.Second) + waitForNothing("retry cancelled", t, timer, obj) + + // Run, queue a retry from a goroutine + obj.setRetryFn(func() { + go func() { + time.Sleep(100 * time.Millisecond) + runner.RetryAfter(5 * time.Second) + }() + }) + runner.Run() + waitForRunWithRetry("fourth run", t, timer, obj, 5*time.Second) + + // Call Run again before minInterval passes + timer.advance(100 * time.Millisecond) // rel=100ms + runner.Run() + waitForDefer("too soon after fourth run", t, timer, obj, 900*time.Millisecond) + + // Deferred run will run after minInterval passes + timer.advance(900 * time.Millisecond) // rel=1000ms + waitForRun("fifth run", t, timer, obj) + + // Retry was cancelled because we already ran + timer.advance(4 * time.Second) // rel=4s since run, 5s since RetryAfter + waitForNothing("retry cancelled", t, timer, obj) + + // Rerun happens after maxInterval + timer.advance(5 * time.Second) // rel=9s since run, 10s since RetryAfter + waitForNothing("premature", t, timer, obj) + timer.advance(time.Second) // rel=10s since run + waitForRun("maxInterval", t, timer, obj) + + // Clean up. + stop <- struct{}{} + // a message is sent to time.updated in func Stop() at the end of the child goroutine + // to terminate the child, a receive on time.updated is needed here + <-timer.updated +} diff --git a/pkg/proxy/winkernel/proxier.go b/pkg/proxy/winkernel/proxier.go index 91fb6fad8d4..14880aed708 100644 --- a/pkg/proxy/winkernel/proxier.go +++ b/pkg/proxy/winkernel/proxier.go @@ -48,8 +48,8 @@ import ( "k8s.io/kubernetes/pkg/proxy/healthcheck" "k8s.io/kubernetes/pkg/proxy/metaproxier" "k8s.io/kubernetes/pkg/proxy/metrics" + "k8s.io/kubernetes/pkg/proxy/runner" proxyutil "k8s.io/kubernetes/pkg/proxy/util" - "k8s.io/kubernetes/pkg/util/async" netutils "k8s.io/utils/net" ) @@ -660,7 +660,7 @@ type Proxier struct { endpointSlicesSynced bool servicesSynced bool initialized int32 - syncRunner *async.BoundedFrequencyRunner // governs calls to syncProxyRules + syncRunner *runner.BoundedFrequencyRunner // governs calls to syncProxyRules // These are effectively const and do not need the mutex to be held. nodeName string nodeIP net.IP @@ -755,7 +755,7 @@ func NewProxier( burstSyncs := 2 klog.V(3).InfoS("Record sync param", "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod, "burstSyncs", burstSyncs) - proxier.syncRunner = async.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, burstSyncs) + proxier.syncRunner = runner.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, burstSyncs) return proxier, nil } From 0298e04ea0fa1f47db650886d08ae47cdab05985 Mon Sep 17 00:00:00 2001 From: Dan Winship Date: Mon, 5 May 2025 14:53:03 -0400 Subject: [PATCH 2/5] Updates to BoundedFrequencyRunner - Use structured logging. - Use t.Helper() in unit tests. - Improve some comments. - Remove an unnecessary check/panic. Co-authored-by: Antonio Ojea --- pkg/proxy/runner/bounded_frequency_runner.go | 65 +++++++------------ .../runner/bounded_frequency_runner_test.go | 7 ++ 2 files changed, 30 insertions(+), 42 deletions(-) diff --git a/pkg/proxy/runner/bounded_frequency_runner.go b/pkg/proxy/runner/bounded_frequency_runner.go index 8b13c2fd4b5..3ad3255e4b0 100644 --- a/pkg/proxy/runner/bounded_frequency_runner.go +++ b/pkg/proxy/runner/bounded_frequency_runner.go @@ -26,8 +26,7 @@ import ( "k8s.io/klog/v2" ) -// BoundedFrequencyRunner manages runs of a user-provided function. -// See NewBoundedFrequencyRunner for examples. +// BoundedFrequencyRunner manages runs of a user-provided work function. type BoundedFrequencyRunner struct { name string // the name of this instance minInterval time.Duration // the min time between runs, modulo bursts @@ -36,7 +35,7 @@ type BoundedFrequencyRunner struct { run chan struct{} // try an async run mu sync.Mutex // guards runs of fn and all mutations - fn func() // function to run + fn func() // the work function lastRun time.Time // time of last run timer timer // timer for deferred runs limiter rateLimiter // rate limiter for on-demand runs @@ -123,35 +122,24 @@ func (rt *realTimer) Sleep(d time.Duration) { var _ timer = &realTimer{} -// NewBoundedFrequencyRunner creates a new BoundedFrequencyRunner instance, -// which will manage runs of the specified function. +// NewBoundedFrequencyRunner creates and returns a new BoundedFrequencyRunner. +// This runner manages the execution frequency of the provided work function `fn`. // // All runs will be async to the caller of BoundedFrequencyRunner.Run, but // multiple runs are serialized. If the function needs to hold locks, it must // take them internally. // -// Runs of the function will have at least minInterval between them (from -// completion to next start), except that up to bursts may be allowed. Burst -// runs are "accumulated" over time, one per minInterval up to burstRuns total. -// This can be used, for example, to mitigate the impact of expensive operations -// being called in response to user-initiated operations. Run requests that -// would violate the minInterval are coalesced and run at the next opportunity. +// The runner guarantees two properties: +// 1. Minimum Interval (`minInterval`): At least `minInterval` must pass between +// the *completion* of one execution and the *start* of the next. Calls to +// `Run()` during this cooldown period are coalesced and deferred until the +// interval expires. This prevents burst executions. +// 2. Maximum Interval (`maxInterval`): The function `fn` is guaranteed to run +// at least once per `maxInterval`, ensuring periodic execution even without +// explicit `Run()` calls (e.g., for refreshing state). // -// The function will be run at least once per maxInterval. For example, this can -// force periodic refreshes of state in the absence of anyone calling Run. -// -// Examples: -// -// NewBoundedFrequencyRunner("name", fn, time.Second, 5*time.Second, 1) -// - fn will have at least 1 second between runs -// - fn will have no more than 5 seconds between runs -// -// NewBoundedFrequencyRunner("name", fn, 3*time.Second, 10*time.Second, 3) -// - fn will have at least 3 seconds between runs, with up to 3 burst runs -// - fn will have no more than 10 seconds between runs -// -// The maxInterval must be greater than or equal to the minInterval, If the -// caller passes a maxInterval less than minInterval, this function will panic. +// `maxInterval` must be greater than or equal to `minInterval`; otherwise, +// this function will panic. func NewBoundedFrequencyRunner(name string, fn func(), minInterval, maxInterval time.Duration, burstRuns int) *BoundedFrequencyRunner { timer := &realTimer{timer: time.NewTimer(0)} // will tick immediately <-timer.C() // consume the first tick @@ -163,9 +151,6 @@ func construct(name string, fn func(), minInterval, maxInterval time.Duration, b if maxInterval < minInterval { panic(fmt.Sprintf("%s: maxInterval (%v) must be >= minInterval (%v)", name, maxInterval, minInterval)) } - if timer == nil { - panic(fmt.Sprintf("%s: timer must be non-nil", name)) - } bfr := &BoundedFrequencyRunner{ name: name, @@ -189,13 +174,13 @@ func construct(name string, fn func(), minInterval, maxInterval time.Duration, b // Loop handles the periodic timer and run requests. This is expected to be // called as a goroutine. func (bfr *BoundedFrequencyRunner) Loop(stop <-chan struct{}) { - klog.V(3).Infof("%s Loop running", bfr.name) + klog.V(3).InfoS("Loop running", "runner", bfr.name) bfr.timer.Reset(bfr.maxInterval) for { select { case <-stop: bfr.stop() - klog.V(3).Infof("%s Loop stopping", bfr.name) + klog.V(3).InfoS("Loop stopping", "runner", bfr.name) return case <-bfr.timer.C(): bfr.tryRun() @@ -207,16 +192,12 @@ func (bfr *BoundedFrequencyRunner) Loop(stop <-chan struct{}) { } } -// Run the function as soon as possible. If this is called while Loop is not +// Run the work function as soon as possible. If this is called while Loop is not // running, the call may be deferred indefinitely. -// If there is already a queued request to call the underlying function, it -// may be dropped - it is just guaranteed that we will try calling the -// underlying function as soon as possible starting from now. +// Once there is a queued request to call the work function, further calls to +// Run() will have no effect until after it runs. func (bfr *BoundedFrequencyRunner) Run() { - // If it takes a lot of time to run the underlying function, noone is really - // processing elements from channel. So to avoid blocking here on the - // putting element to it, we simply skip it if there is already an element - // in it. + // If bfr.run is empty, push an element onto it. Otherwise, do nothing. select { case bfr.run <- struct{}{}: default: @@ -276,7 +257,7 @@ func (bfr *BoundedFrequencyRunner) doRetry() { retryInterval := bfr.retryTime.Sub(bfr.timer.Now()) bfr.retryTime = time.Time{} if retryInterval < bfr.timer.Remaining() { - klog.V(3).Infof("%s: retrying in %v", bfr.name, retryInterval) + klog.V(3).InfoS("retrying", "runner", bfr.name, "interval", retryInterval) bfr.timer.Stop() bfr.timer.Reset(retryInterval) } @@ -293,7 +274,7 @@ func (bfr *BoundedFrequencyRunner) tryRun() { bfr.lastRun = bfr.timer.Now() bfr.timer.Stop() bfr.timer.Reset(bfr.maxInterval) - klog.V(3).Infof("%s: ran, next possible in %v, periodic in %v", bfr.name, bfr.minInterval, bfr.maxInterval) + klog.V(3).InfoS("ran", "runner", bfr.name, "minInterval", bfr.minInterval, "maxInternval", bfr.maxInterval) return } @@ -301,7 +282,7 @@ func (bfr *BoundedFrequencyRunner) tryRun() { elapsed := bfr.timer.Since(bfr.lastRun) // how long since last run nextPossible := bfr.minInterval - elapsed // time to next possible run nextScheduled := bfr.timer.Remaining() // time to next scheduled run - klog.V(4).Infof("%s: %v since last run, possible in %v, scheduled in %v", bfr.name, elapsed, nextPossible, nextScheduled) + klog.V(4).InfoS("can't run", "runner", bfr.name, "elapsed", elapsed, "nextPossible", nextPossible, "nextScheduled", nextScheduled) // It's hard to avoid race conditions in the unit tests unless we always reset // the timer here, even when it's unchanged diff --git a/pkg/proxy/runner/bounded_frequency_runner_test.go b/pkg/proxy/runner/bounded_frequency_runner_test.go index e4f4e4933d2..3d912f15249 100644 --- a/pkg/proxy/runner/bounded_frequency_runner_test.go +++ b/pkg/proxy/runner/bounded_frequency_runner_test.go @@ -152,6 +152,7 @@ func (ft *fakeTimer) advance(d time.Duration) { // return the calling line number (for printing) // test the timer's state func checkTimer(name string, t *testing.T, upd timerUpdate, active bool, next time.Duration) { + t.Helper() if upd.active != active { t.Fatalf("%s: expected timer active=%v", name, active) } @@ -162,6 +163,7 @@ func checkTimer(name string, t *testing.T, upd timerUpdate, active bool, next ti // test and reset the receiver's state func checkReceiver(name string, t *testing.T, receiver *receiver, expected bool) { + t.Helper() triggered := receiver.reset() if expected && !triggered { t.Fatalf("%s: function should have been called", name) @@ -175,6 +177,7 @@ var minInterval = 1 * time.Second var maxInterval = 10 * time.Second func waitForReset(name string, t *testing.T, timer *fakeTimer, obj *receiver, expectCall bool, expectNext time.Duration) { + t.Helper() upd := <-timer.updated // wait for stop checkReceiver(name, t, obj, expectCall) checkReceiver(name, t, obj, false) // prove post-condition @@ -184,20 +187,24 @@ func waitForReset(name string, t *testing.T, timer *fakeTimer, obj *receiver, ex } func waitForRun(name string, t *testing.T, timer *fakeTimer, obj *receiver) { + t.Helper() waitForReset(name, t, timer, obj, true, maxInterval) } func waitForRunWithRetry(name string, t *testing.T, timer *fakeTimer, obj *receiver, expectNext time.Duration) { + t.Helper() // It will first get reset as with a normal run, and then get set again waitForRun(name, t, timer, obj) waitForReset(name, t, timer, obj, false, expectNext) } func waitForDefer(name string, t *testing.T, timer *fakeTimer, obj *receiver, expectNext time.Duration) { + t.Helper() waitForReset(name, t, timer, obj, false, expectNext) } func waitForNothing(name string, t *testing.T, timer *fakeTimer, obj *receiver) { + t.Helper() select { case <-timer.c: t.Fatalf("%s: unexpected timer tick", name) From c16ee887efc30e956e23331cbdc8e48b4530ca54 Mon Sep 17 00:00:00 2001 From: Dan Winship Date: Mon, 5 May 2025 14:26:28 -0400 Subject: [PATCH 3/5] Remove burst syncs from BoundedFrequencyRunner Burst syncs are theoretically useful for dealing with a single change that results in multiple Run() calls (eg, a Service and EndpointSlice both changing), but 2 isn't enough to cover all cases, and a better way of dealing with this problem is to just use a smaller minSyncPeriod. Co-authored-by: Antonio Ojea --- pkg/proxy/iptables/proxier.go | 5 +- pkg/proxy/iptables/proxier_test.go | 2 +- pkg/proxy/ipvs/proxier.go | 5 +- pkg/proxy/ipvs/proxier_test.go | 2 +- pkg/proxy/nftables/proxier.go | 5 +- pkg/proxy/nftables/proxier_test.go | 2 +- pkg/proxy/runner/bounded_frequency_runner.go | 11 ++- .../runner/bounded_frequency_runner_test.go | 90 +------------------ pkg/proxy/winkernel/proxier.go | 5 +- 9 files changed, 19 insertions(+), 108 deletions(-) diff --git a/pkg/proxy/iptables/proxier.go b/pkg/proxy/iptables/proxier.go index 600ab9b62bf..aa663139bd7 100644 --- a/pkg/proxy/iptables/proxier.go +++ b/pkg/proxy/iptables/proxier.go @@ -307,12 +307,11 @@ func NewProxier(ctx context.Context, }, } - burstSyncs := 2 - logger.V(2).Info("Iptables sync params", "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod, "burstSyncs", burstSyncs) + logger.V(2).Info("Iptables sync params", "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod) // We pass syncPeriod to ipt.Monitor, which will call us only if it needs to. // We need to pass *some* maxInterval to NewBoundedFrequencyRunner anyway though. // time.Hour is arbitrary. - proxier.syncRunner = runner.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, proxyutil.FullSyncPeriod, burstSyncs) + proxier.syncRunner = runner.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, proxyutil.FullSyncPeriod) go ipt.Monitor(kubeProxyCanaryChain, []utiliptables.Table{utiliptables.TableMangle, utiliptables.TableNAT, utiliptables.TableFilter}, proxier.forceSyncProxyRules, syncPeriod, wait.NeverStop) diff --git a/pkg/proxy/iptables/proxier_test.go b/pkg/proxy/iptables/proxier_test.go index 430d7c34a77..5d1f41423d6 100644 --- a/pkg/proxy/iptables/proxier_test.go +++ b/pkg/proxy/iptables/proxier_test.go @@ -144,7 +144,7 @@ func NewFakeProxier(ipt utiliptables.Interface) *Proxier { }, } p.setInitialized(true) - p.syncRunner = runner.NewBoundedFrequencyRunner("test-sync-runner", p.syncProxyRules, 0, time.Minute, 1) + p.syncRunner = runner.NewBoundedFrequencyRunner("test-sync-runner", p.syncProxyRules, 0, time.Minute) return p } diff --git a/pkg/proxy/ipvs/proxier.go b/pkg/proxy/ipvs/proxier.go index 7995730d320..eb342082a7f 100644 --- a/pkg/proxy/ipvs/proxier.go +++ b/pkg/proxy/ipvs/proxier.go @@ -400,9 +400,8 @@ func NewProxier( for _, is := range ipsetInfo { proxier.ipsetList[is.name] = NewIPSet(ipset, is.name, is.setType, (ipFamily == v1.IPv6Protocol), is.comment) } - burstSyncs := 2 - logger.V(2).Info("ipvs sync params", "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod, "burstSyncs", burstSyncs) - proxier.syncRunner = runner.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, burstSyncs) + logger.V(2).Info("ipvs sync params", "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod) + proxier.syncRunner = runner.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod) proxier.gracefuldeleteManager.Run() return proxier, nil } diff --git a/pkg/proxy/ipvs/proxier_test.go b/pkg/proxy/ipvs/proxier_test.go index 5befa4d2476..df5f6c3620e 100644 --- a/pkg/proxy/ipvs/proxier_test.go +++ b/pkg/proxy/ipvs/proxier_test.go @@ -167,7 +167,7 @@ func NewFakeProxier(ctx context.Context, ipt utiliptables.Interface, ipvs utilip ipFamily: ipFamily, } p.setInitialized(true) - p.syncRunner = runner.NewBoundedFrequencyRunner("test-sync-runner", p.syncProxyRules, 0, time.Minute, 1) + p.syncRunner = runner.NewBoundedFrequencyRunner("test-sync-runner", p.syncProxyRules, 0, time.Minute) return p } diff --git a/pkg/proxy/nftables/proxier.go b/pkg/proxy/nftables/proxier.go index 6e2ce149fb3..7947405672b 100644 --- a/pkg/proxy/nftables/proxier.go +++ b/pkg/proxy/nftables/proxier.go @@ -273,10 +273,9 @@ func NewProxier(ctx context.Context, serviceNodePorts: newNFTElementStorage("map", serviceNodePortsMap), } - burstSyncs := 2 - logger.V(2).Info("NFTables sync params", "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod, "burstSyncs", burstSyncs) + logger.V(2).Info("NFTables sync params", "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod) // We need to pass *some* maxInterval to NewBoundedFrequencyRunner. time.Hour is arbitrary. - proxier.syncRunner = runner.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, proxyutil.FullSyncPeriod, burstSyncs) + proxier.syncRunner = runner.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, proxyutil.FullSyncPeriod) return proxier, nil } diff --git a/pkg/proxy/nftables/proxier_test.go b/pkg/proxy/nftables/proxier_test.go index cc19185639e..ecd0fd9b45c 100644 --- a/pkg/proxy/nftables/proxier_test.go +++ b/pkg/proxy/nftables/proxier_test.go @@ -142,7 +142,7 @@ func NewFakeProxier(ipFamily v1.IPFamily) (*knftables.Fake, *Proxier) { serviceNodePorts: newNFTElementStorage("map", serviceNodePortsMap), } p.setInitialized(true) - p.syncRunner = runner.NewBoundedFrequencyRunner("test-sync-runner", p.syncProxyRules, 0, time.Minute, 1) + p.syncRunner = runner.NewBoundedFrequencyRunner("test-sync-runner", p.syncProxyRules, 0, time.Minute) return nft, p } diff --git a/pkg/proxy/runner/bounded_frequency_runner.go b/pkg/proxy/runner/bounded_frequency_runner.go index 3ad3255e4b0..6bfbb3e5f7b 100644 --- a/pkg/proxy/runner/bounded_frequency_runner.go +++ b/pkg/proxy/runner/bounded_frequency_runner.go @@ -29,7 +29,7 @@ import ( // BoundedFrequencyRunner manages runs of a user-provided work function. type BoundedFrequencyRunner struct { name string // the name of this instance - minInterval time.Duration // the min time between runs, modulo bursts + minInterval time.Duration // the min time between runs maxInterval time.Duration // the max time between runs run chan struct{} // try an async run @@ -140,14 +140,14 @@ var _ timer = &realTimer{} // // `maxInterval` must be greater than or equal to `minInterval`; otherwise, // this function will panic. -func NewBoundedFrequencyRunner(name string, fn func(), minInterval, maxInterval time.Duration, burstRuns int) *BoundedFrequencyRunner { +func NewBoundedFrequencyRunner(name string, fn func(), minInterval, maxInterval time.Duration) *BoundedFrequencyRunner { timer := &realTimer{timer: time.NewTimer(0)} // will tick immediately <-timer.C() // consume the first tick - return construct(name, fn, minInterval, maxInterval, burstRuns, timer) + return construct(name, fn, minInterval, maxInterval, timer) } // Make an instance with dependencies injected. -func construct(name string, fn func(), minInterval, maxInterval time.Duration, burstRuns int, timer timer) *BoundedFrequencyRunner { +func construct(name string, fn func(), minInterval, maxInterval time.Duration, timer timer) *BoundedFrequencyRunner { if maxInterval < minInterval { panic(fmt.Sprintf("%s: maxInterval (%v) must be >= minInterval (%v)", name, maxInterval, minInterval)) } @@ -164,9 +164,8 @@ func construct(name string, fn func(), minInterval, maxInterval time.Duration, b if minInterval == 0 { bfr.limiter = nullLimiter{} } else { - // allow burst updates in short succession qps := float32(time.Second) / float32(minInterval) - bfr.limiter = flowcontrol.NewTokenBucketRateLimiterWithClock(qps, burstRuns, timer) + bfr.limiter = flowcontrol.NewTokenBucketRateLimiterWithClock(qps, 1, timer) } return bfr } diff --git a/pkg/proxy/runner/bounded_frequency_runner_test.go b/pkg/proxy/runner/bounded_frequency_runner_test.go index 3d912f15249..12f9ba08303 100644 --- a/pkg/proxy/runner/bounded_frequency_runner_test.go +++ b/pkg/proxy/runner/bounded_frequency_runner_test.go @@ -215,10 +215,10 @@ func waitForNothing(name string, t *testing.T, timer *fakeTimer, obj *receiver) checkReceiver(name, t, obj, false) } -func Test_BoundedFrequencyRunnerNoBurst(t *testing.T) { +func Test_BoundedFrequencyRunner(t *testing.T) { obj := &receiver{} timer := newFakeTimer() - runner := construct("test-runner", obj.F, minInterval, maxInterval, 1, timer) + runner := construct("test-runner", obj.F, minInterval, maxInterval, timer) stop := make(chan struct{}) var upd timerUpdate @@ -289,94 +289,10 @@ func Test_BoundedFrequencyRunnerNoBurst(t *testing.T) { <-timer.updated } -func Test_BoundedFrequencyRunnerBurst(t *testing.T) { - obj := &receiver{} - timer := newFakeTimer() - runner := construct("test-runner", obj.F, minInterval, maxInterval, 2, timer) - stop := make(chan struct{}) - - var upd timerUpdate - - // Start. - go runner.Loop(stop) - upd = <-timer.updated // wait for initial time to be set to max - checkTimer("init", t, upd, true, maxInterval) - checkReceiver("init", t, obj, false) - - // Run once, immediately. - // abs=0ms, rel=0ms - runner.Run() - waitForRun("first run", t, timer, obj) - - // Run again, before minInterval expires, with burst. - timer.advance(1 * time.Millisecond) // abs=1ms, rel=1ms - runner.Run() - waitForRun("second run", t, timer, obj) - - // Run again, before minInterval expires. - timer.advance(498 * time.Millisecond) // abs=499ms, rel=498ms - runner.Run() - waitForDefer("too soon after second", t, timer, obj, 502*time.Millisecond) - - // Run again, before minInterval expires. - timer.advance(1 * time.Millisecond) // abs=500ms, rel=499ms - runner.Run() - waitForDefer("too soon after second 2", t, timer, obj, 501*time.Millisecond) - - // Run again, before minInterval expires. - timer.advance(1 * time.Millisecond) // abs=501ms, rel=500ms - runner.Run() - waitForDefer("too soon after second 3", t, timer, obj, 500*time.Millisecond) - - // Advance timer enough to replenish bursts, but not enough to be minInterval - // after the last run - timer.advance(499 * time.Millisecond) // abs=1000ms, rel=999ms - waitForNothing("not minInterval", t, timer, obj) - runner.Run() - waitForRun("third run", t, timer, obj) - - // Run again, before minInterval expires. - timer.advance(1 * time.Millisecond) // abs=1001ms, rel=1ms - runner.Run() - waitForDefer("too soon after third", t, timer, obj, 999*time.Millisecond) - - // Run again, before minInterval expires. - timer.advance(998 * time.Millisecond) // abs=1999ms, rel=999ms - runner.Run() - waitForDefer("too soon after third 2", t, timer, obj, 1*time.Millisecond) - - // Advance and do the deferred run - timer.advance(1 * time.Millisecond) // abs=2000ms, rel=1000ms - waitForRun("fourth run", t, timer, obj) - - // Run again, once burst has fully replenished. - timer.advance(2 * time.Second) // abs=4000ms, rel=2000ms - runner.Run() - waitForRun("fifth run", t, timer, obj) - runner.Run() - waitForRun("sixth run", t, timer, obj) - runner.Run() - waitForDefer("too soon after sixth", t, timer, obj, 1*time.Second) - - // Wait until minInterval after the last run - timer.advance(1 * time.Second) // abs=5000ms, rel=1000ms - waitForRun("seventh run", t, timer, obj) - - // Wait for maxInterval - timer.advance(10 * time.Second) // abs=15000ms, rel=10000ms - waitForRun("maxInterval", t, timer, obj) - - // Clean up. - stop <- struct{}{} - // a message is sent to time.updated in func Stop() at the end of the child goroutine - // to terminate the child, a receive on time.updated is needed here - <-timer.updated -} - func Test_BoundedFrequencyRunnerRetryAfter(t *testing.T) { obj := &receiver{} timer := newFakeTimer() - runner := construct("test-runner", obj.F, minInterval, maxInterval, 1, timer) + runner := construct("test-runner", obj.F, minInterval, maxInterval, timer) stop := make(chan struct{}) var upd timerUpdate diff --git a/pkg/proxy/winkernel/proxier.go b/pkg/proxy/winkernel/proxier.go index 14880aed708..bff18030b32 100644 --- a/pkg/proxy/winkernel/proxier.go +++ b/pkg/proxy/winkernel/proxier.go @@ -753,9 +753,8 @@ func NewProxier( return nil, err } - burstSyncs := 2 - klog.V(3).InfoS("Record sync param", "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod, "burstSyncs", burstSyncs) - proxier.syncRunner = runner.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, burstSyncs) + klog.V(3).InfoS("Record sync param", "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod) + proxier.syncRunner = runner.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod) return proxier, nil } From eae17c21b02f20c0b630c9e8d4f9d7215a70fe39 Mon Sep 17 00:00:00 2001 From: Dan Winship Date: Mon, 5 May 2025 14:26:28 -0400 Subject: [PATCH 4/5] Change how BoundedFrequencyRunner retries work Rather than having a RetryAfter function, do a retry (at a fixed interval) if the work function returns an error. Co-authored-by: Antonio Ojea --- pkg/proxy/iptables/proxier.go | 9 +- pkg/proxy/iptables/proxier_test.go | 2 +- pkg/proxy/ipvs/proxier.go | 9 +- pkg/proxy/ipvs/proxier_test.go | 2 +- pkg/proxy/nftables/proxier.go | 9 +- pkg/proxy/nftables/proxier_test.go | 2 +- pkg/proxy/runner/bounded_frequency_runner.go | 115 ++++++------------ .../runner/bounded_frequency_runner_test.go | 46 ++++--- pkg/proxy/winkernel/proxier.go | 8 +- 9 files changed, 82 insertions(+), 120 deletions(-) diff --git a/pkg/proxy/iptables/proxier.go b/pkg/proxy/iptables/proxier.go index aa663139bd7..4fcdc79bca3 100644 --- a/pkg/proxy/iptables/proxier.go +++ b/pkg/proxy/iptables/proxier.go @@ -307,11 +307,11 @@ func NewProxier(ctx context.Context, }, } - logger.V(2).Info("Iptables sync params", "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod) + logger.V(2).Info("Iptables sync params", "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod, "maxSyncPeriod", proxyutil.FullSyncPeriod) // We pass syncPeriod to ipt.Monitor, which will call us only if it needs to. // We need to pass *some* maxInterval to NewBoundedFrequencyRunner anyway though. // time.Hour is arbitrary. - proxier.syncRunner = runner.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, proxyutil.FullSyncPeriod) + proxier.syncRunner = runner.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, proxyutil.FullSyncPeriod) go ipt.Monitor(kubeProxyCanaryChain, []utiliptables.Table{utiliptables.TableMangle, utiliptables.TableNAT, utiliptables.TableFilter}, proxier.forceSyncProxyRules, syncPeriod, wait.NeverStop) @@ -783,7 +783,7 @@ func (proxier *Proxier) forceSyncProxyRules() { // This is where all of the iptables-save/restore calls happen. // The only other iptables rules are those that are setup in iptablesInit() // This assumes proxier.mu is NOT held -func (proxier *Proxier) syncProxyRules() { +func (proxier *Proxier) syncProxyRules() (retryError error) { proxier.mu.Lock() defer proxier.mu.Unlock() @@ -817,7 +817,7 @@ func (proxier *Proxier) syncProxyRules() { defer func() { if !success { proxier.logger.Info("Sync failed", "retryingTime", proxier.syncPeriod) - proxier.syncRunner.RetryAfter(proxier.syncPeriod) + retryError = fmt.Errorf("Sync failed") if !doFullSync { metrics.IPTablesPartialRestoreFailuresTotal.WithLabelValues(string(proxier.ipFamily)).Inc() } @@ -1586,6 +1586,7 @@ func (proxier *Proxier) syncProxyRules() { // Finish housekeeping, clear stale conntrack entries for UDP Services conntrack.CleanStaleEntries(proxier.conntrack, proxier.ipFamily, proxier.svcPortMap, proxier.endpointsMap) } + return } func (proxier *Proxier) writeServiceToEndpointRules(natRules proxyutil.LineBuffer, svcPortNameString string, svcInfo proxy.ServicePort, svcChain utiliptables.Chain, endpoints []proxy.Endpoint, args []string) { diff --git a/pkg/proxy/iptables/proxier_test.go b/pkg/proxy/iptables/proxier_test.go index 5d1f41423d6..85f11b241e1 100644 --- a/pkg/proxy/iptables/proxier_test.go +++ b/pkg/proxy/iptables/proxier_test.go @@ -144,7 +144,7 @@ func NewFakeProxier(ipt utiliptables.Interface) *Proxier { }, } p.setInitialized(true) - p.syncRunner = runner.NewBoundedFrequencyRunner("test-sync-runner", p.syncProxyRules, 0, time.Minute) + p.syncRunner = runner.NewBoundedFrequencyRunner("test-sync-runner", p.syncProxyRules, 0, 30*time.Second, time.Minute) return p } diff --git a/pkg/proxy/ipvs/proxier.go b/pkg/proxy/ipvs/proxier.go index eb342082a7f..1bdd36a1af5 100644 --- a/pkg/proxy/ipvs/proxier.go +++ b/pkg/proxy/ipvs/proxier.go @@ -400,8 +400,10 @@ func NewProxier( for _, is := range ipsetInfo { proxier.ipsetList[is.name] = NewIPSet(ipset, is.name, is.setType, (ipFamily == v1.IPv6Protocol), is.comment) } - logger.V(2).Info("ipvs sync params", "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod) - proxier.syncRunner = runner.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod) + + logger.V(2).Info("ipvs sync params", "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod, "maxSyncPeriod", proxyutil.FullSyncPeriod) + proxier.syncRunner = runner.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, proxyutil.FullSyncPeriod) + proxier.gracefuldeleteManager.Run() return proxier, nil } @@ -893,7 +895,7 @@ func (proxier *Proxier) OnNodeSynced() { func (proxier *Proxier) OnServiceCIDRsChanged(_ []string) {} // This is where all of the ipvs calls happen. -func (proxier *Proxier) syncProxyRules() { +func (proxier *Proxier) syncProxyRules() (retryError error) { proxier.mu.Lock() defer proxier.mu.Unlock() @@ -1488,6 +1490,7 @@ func (proxier *Proxier) syncProxyRules() { // Finish housekeeping, clear stale conntrack entries for UDP Services conntrack.CleanStaleEntries(proxier.conntrack, proxier.ipFamily, proxier.svcPortMap, proxier.endpointsMap) } + return } // writeIptablesRules write all iptables rules to proxier.natRules or proxier.FilterRules that ipvs proxier needed diff --git a/pkg/proxy/ipvs/proxier_test.go b/pkg/proxy/ipvs/proxier_test.go index df5f6c3620e..3d6608bbaae 100644 --- a/pkg/proxy/ipvs/proxier_test.go +++ b/pkg/proxy/ipvs/proxier_test.go @@ -167,7 +167,7 @@ func NewFakeProxier(ctx context.Context, ipt utiliptables.Interface, ipvs utilip ipFamily: ipFamily, } p.setInitialized(true) - p.syncRunner = runner.NewBoundedFrequencyRunner("test-sync-runner", p.syncProxyRules, 0, time.Minute) + p.syncRunner = runner.NewBoundedFrequencyRunner("test-sync-runner", p.syncProxyRules, 0, 30*time.Second, time.Minute) return p } diff --git a/pkg/proxy/nftables/proxier.go b/pkg/proxy/nftables/proxier.go index 7947405672b..3c3ee0199ab 100644 --- a/pkg/proxy/nftables/proxier.go +++ b/pkg/proxy/nftables/proxier.go @@ -273,9 +273,9 @@ func NewProxier(ctx context.Context, serviceNodePorts: newNFTElementStorage("map", serviceNodePortsMap), } - logger.V(2).Info("NFTables sync params", "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod) + logger.V(2).Info("NFTables sync params", "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod, "maxSyncPeriod", proxyutil.FullSyncPeriod) // We need to pass *some* maxInterval to NewBoundedFrequencyRunner. time.Hour is arbitrary. - proxier.syncRunner = runner.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, proxyutil.FullSyncPeriod) + proxier.syncRunner = runner.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, proxyutil.FullSyncPeriod) return proxier, nil } @@ -1161,7 +1161,7 @@ func (proxier *Proxier) logFailure(tx *knftables.Transaction) { // This is where all of the nftables calls happen. // This assumes proxier.mu is NOT held -func (proxier *Proxier) syncProxyRules() { +func (proxier *Proxier) syncProxyRules() (retryError error) { proxier.mu.Lock() defer proxier.mu.Unlock() @@ -1200,7 +1200,7 @@ func (proxier *Proxier) syncProxyRules() { defer func() { if !success { proxier.logger.Info("Sync failed", "retryingTime", proxier.syncPeriod) - proxier.syncRunner.RetryAfter(proxier.syncPeriod) + retryError = fmt.Errorf("Sync failed") // proxier.serviceChanges and proxier.endpointChanges have already // been flushed, so we've lost the state needed to be able to do // a partial sync. @@ -1886,6 +1886,7 @@ func (proxier *Proxier) syncProxyRules() { // Finish housekeeping, clear stale conntrack entries for UDP Services conntrack.CleanStaleEntries(proxier.conntrack, proxier.ipFamily, proxier.svcPortMap, proxier.endpointsMap) } + return } // epChainSkipUpdate returns true if the EP chain doesn't need to be updated. diff --git a/pkg/proxy/nftables/proxier_test.go b/pkg/proxy/nftables/proxier_test.go index ecd0fd9b45c..675ade0c194 100644 --- a/pkg/proxy/nftables/proxier_test.go +++ b/pkg/proxy/nftables/proxier_test.go @@ -142,7 +142,7 @@ func NewFakeProxier(ipFamily v1.IPFamily) (*knftables.Fake, *Proxier) { serviceNodePorts: newNFTElementStorage("map", serviceNodePortsMap), } p.setInitialized(true) - p.syncRunner = runner.NewBoundedFrequencyRunner("test-sync-runner", p.syncProxyRules, 0, time.Minute) + p.syncRunner = runner.NewBoundedFrequencyRunner("test-sync-runner", p.syncProxyRules, 0, 30*time.Second, time.Minute) return nft, p } diff --git a/pkg/proxy/runner/bounded_frequency_runner.go b/pkg/proxy/runner/bounded_frequency_runner.go index 6bfbb3e5f7b..525fa8f3c6d 100644 --- a/pkg/proxy/runner/bounded_frequency_runner.go +++ b/pkg/proxy/runner/bounded_frequency_runner.go @@ -28,21 +28,19 @@ import ( // BoundedFrequencyRunner manages runs of a user-provided work function. type BoundedFrequencyRunner struct { - name string // the name of this instance - minInterval time.Duration // the min time between runs - maxInterval time.Duration // the max time between runs + name string // the name of this instance + + minInterval time.Duration // the min time between runs + retryInterval time.Duration // the time between a run and a retry + maxInterval time.Duration // the max time between runs run chan struct{} // try an async run - mu sync.Mutex // guards runs of fn and all mutations - fn func() // the work function - lastRun time.Time // time of last run - timer timer // timer for deferred runs - limiter rateLimiter // rate limiter for on-demand runs - - retry chan struct{} // schedule a retry - retryMu sync.Mutex // guards retryTime - retryTime time.Time // when to retry + mu sync.Mutex // guards runs of fn and all mutations + fn func() error // the work function + lastRun time.Time // time of last run + timer timer // timer for deferred runs + limiter rateLimiter // rate limiter for on-demand runs } // designed so that flowcontrol.RateLimiter satisfies @@ -140,26 +138,32 @@ var _ timer = &realTimer{} // // `maxInterval` must be greater than or equal to `minInterval`; otherwise, // this function will panic. -func NewBoundedFrequencyRunner(name string, fn func(), minInterval, maxInterval time.Duration) *BoundedFrequencyRunner { +// +// If `fn` returns an error, then it will be run again no later than `retryInterval` +// (unless another trigger, like `Run()` or `maxInterval`, causes it to run sooner). Any +// successful run will abort the retry attempt. +func NewBoundedFrequencyRunner(name string, fn func() error, minInterval, retryInterval, maxInterval time.Duration) *BoundedFrequencyRunner { timer := &realTimer{timer: time.NewTimer(0)} // will tick immediately <-timer.C() // consume the first tick - return construct(name, fn, minInterval, maxInterval, timer) + return construct(name, fn, minInterval, retryInterval, maxInterval, timer) } // Make an instance with dependencies injected. -func construct(name string, fn func(), minInterval, maxInterval time.Duration, timer timer) *BoundedFrequencyRunner { +func construct(name string, fn func() error, minInterval, retryInterval, maxInterval time.Duration, timer timer) *BoundedFrequencyRunner { if maxInterval < minInterval { panic(fmt.Sprintf("%s: maxInterval (%v) must be >= minInterval (%v)", name, maxInterval, minInterval)) } bfr := &BoundedFrequencyRunner{ - name: name, - fn: fn, - minInterval: minInterval, - maxInterval: maxInterval, - run: make(chan struct{}, 1), - retry: make(chan struct{}, 1), - timer: timer, + name: name, + fn: fn, + + minInterval: minInterval, + retryInterval: retryInterval, + maxInterval: maxInterval, + + run: make(chan struct{}, 1), + timer: timer, } if minInterval == 0 { bfr.limiter = nullLimiter{} @@ -185,8 +189,6 @@ func (bfr *BoundedFrequencyRunner) Loop(stop <-chan struct{}) { bfr.tryRun() case <-bfr.run: bfr.tryRun() - case <-bfr.retry: - bfr.doRetry() } } } @@ -203,36 +205,6 @@ func (bfr *BoundedFrequencyRunner) Run() { } } -// RetryAfter ensures that the function will run again after no later than interval. This -// can be called from inside a run of the BoundedFrequencyRunner's function, or -// asynchronously. -func (bfr *BoundedFrequencyRunner) RetryAfter(interval time.Duration) { - // This could be called either with or without bfr.mu held, so we can't grab that - // lock, and therefore we can't update the timer directly. - - // If the Loop thread is currently running fn then it may be a while before it - // processes our retry request. But we want to retry at interval from now, not at - // interval from "whenever doRetry eventually gets called". So we convert to - // absolute time. - retryTime := bfr.timer.Now().Add(interval) - - // We can't just write retryTime to a channel because there could be multiple - // RetryAfter calls before Loop gets a chance to read from the channel. So we - // record the soonest requested retry time in bfr.retryTime and then only signal - // the Loop thread once, just like Run does. - bfr.retryMu.Lock() - defer bfr.retryMu.Unlock() - if !bfr.retryTime.IsZero() && bfr.retryTime.Before(retryTime) { - return - } - bfr.retryTime = retryTime - - select { - case bfr.retry <- struct{}{}: - default: - } -} - // assumes the lock is not held func (bfr *BoundedFrequencyRunner) stop() { bfr.mu.Lock() @@ -241,27 +213,6 @@ func (bfr *BoundedFrequencyRunner) stop() { bfr.timer.Stop() } -// assumes the lock is not held -func (bfr *BoundedFrequencyRunner) doRetry() { - bfr.mu.Lock() - defer bfr.mu.Unlock() - bfr.retryMu.Lock() - defer bfr.retryMu.Unlock() - - if bfr.retryTime.IsZero() { - return - } - - // Timer wants an interval not an absolute time, so convert retryTime back now - retryInterval := bfr.retryTime.Sub(bfr.timer.Now()) - bfr.retryTime = time.Time{} - if retryInterval < bfr.timer.Remaining() { - klog.V(3).InfoS("retrying", "runner", bfr.name, "interval", retryInterval) - bfr.timer.Stop() - bfr.timer.Reset(retryInterval) - } -} - // assumes the lock is not held func (bfr *BoundedFrequencyRunner) tryRun() { bfr.mu.Lock() @@ -269,11 +220,19 @@ func (bfr *BoundedFrequencyRunner) tryRun() { if bfr.limiter.TryAccept() { // We're allowed to run the function right now. - bfr.fn() + err := bfr.fn() + bfr.lastRun = bfr.timer.Now() bfr.timer.Stop() - bfr.timer.Reset(bfr.maxInterval) - klog.V(3).InfoS("ran", "runner", bfr.name, "minInterval", bfr.minInterval, "maxInternval", bfr.maxInterval) + + nextInterval := bfr.maxInterval + if err != nil { + // an error will schedule a retry after the retryInterval, + // any successful run before that will stop the retry attempt. + nextInterval = bfr.retryInterval + klog.V(3).InfoS("scheduling retry", "runner", bfr.name, "interval", nextInterval, "error", err) + } + bfr.timer.Reset(nextInterval) return } diff --git a/pkg/proxy/runner/bounded_frequency_runner_test.go b/pkg/proxy/runner/bounded_frequency_runner_test.go index 12f9ba08303..5c7ea48d4f0 100644 --- a/pkg/proxy/runner/bounded_frequency_runner_test.go +++ b/pkg/proxy/runner/bounded_frequency_runner_test.go @@ -17,6 +17,7 @@ limitations under the License. package runner import ( + "fmt" "sync" "testing" "time" @@ -24,20 +25,21 @@ import ( // Track calls to the managed function. type receiver struct { - lock sync.Mutex - run bool - retryFn func() + lock sync.Mutex + run bool + retry bool } -func (r *receiver) F() { +func (r *receiver) F() error { r.lock.Lock() defer r.lock.Unlock() r.run = true - if r.retryFn != nil { - r.retryFn() - r.retryFn = nil + if r.retry { + r.retry = false + return fmt.Errorf("retry") } + return nil } func (r *receiver) reset() bool { @@ -48,10 +50,10 @@ func (r *receiver) reset() bool { return was } -func (r *receiver) setRetryFn(retryFn func()) { +func (r *receiver) setRetry(retry bool) { r.lock.Lock() defer r.lock.Unlock() - r.retryFn = retryFn + r.retry = retry } // A single change event in the fake timer. @@ -174,6 +176,7 @@ func checkReceiver(name string, t *testing.T, receiver *receiver, expected bool) // Durations embedded in test cases depend on these. var minInterval = 1 * time.Second +var retryInterval = 5 * time.Second var maxInterval = 10 * time.Second func waitForReset(name string, t *testing.T, timer *fakeTimer, obj *receiver, expectCall bool, expectNext time.Duration) { @@ -193,9 +196,7 @@ func waitForRun(name string, t *testing.T, timer *fakeTimer, obj *receiver) { func waitForRunWithRetry(name string, t *testing.T, timer *fakeTimer, obj *receiver, expectNext time.Duration) { t.Helper() - // It will first get reset as with a normal run, and then get set again - waitForRun(name, t, timer, obj) - waitForReset(name, t, timer, obj, false, expectNext) + waitForReset(name, t, timer, obj, true, expectNext) } func waitForDefer(name string, t *testing.T, timer *fakeTimer, obj *receiver, expectNext time.Duration) { @@ -218,7 +219,7 @@ func waitForNothing(name string, t *testing.T, timer *fakeTimer, obj *receiver) func Test_BoundedFrequencyRunner(t *testing.T) { obj := &receiver{} timer := newFakeTimer() - runner := construct("test-runner", obj.F, minInterval, maxInterval, timer) + runner := construct("test-runner", obj.F, minInterval, retryInterval, maxInterval, timer) stop := make(chan struct{}) var upd timerUpdate @@ -289,10 +290,10 @@ func Test_BoundedFrequencyRunner(t *testing.T) { <-timer.updated } -func Test_BoundedFrequencyRunnerRetryAfter(t *testing.T) { +func Test_BoundedFrequencyRunnerRetry(t *testing.T) { obj := &receiver{} timer := newFakeTimer() - runner := construct("test-runner", obj.F, minInterval, maxInterval, timer) + runner := construct("test-runner", obj.F, minInterval, retryInterval, maxInterval, timer) stop := make(chan struct{}) var upd timerUpdate @@ -305,7 +306,7 @@ func Test_BoundedFrequencyRunnerRetryAfter(t *testing.T) { // Run once, immediately, and queue a retry // rel=0ms - obj.setRetryFn(func() { runner.RetryAfter(5 * time.Second) }) + obj.setRetry(true) runner.Run() waitForRunWithRetry("first run", t, timer, obj, 5*time.Second) @@ -322,9 +323,9 @@ func Test_BoundedFrequencyRunnerRetryAfter(t *testing.T) { runner.Run() waitForDefer("too soon after retry", t, timer, obj, 501*time.Millisecond) - // Do the deferred run, queue another retry after it returns + // Do the deferred run, have it queue another retry + obj.setRetry(true) timer.advance(501 * time.Millisecond) // rel=1000ms - runner.RetryAfter(5 * time.Second) waitForRunWithRetry("second run", t, timer, obj, 5*time.Second) // Wait for minInterval to pass @@ -339,13 +340,8 @@ func Test_BoundedFrequencyRunnerRetryAfter(t *testing.T) { timer.advance(4 * time.Second) waitForNothing("retry cancelled", t, timer, obj) - // Run, queue a retry from a goroutine - obj.setRetryFn(func() { - go func() { - time.Sleep(100 * time.Millisecond) - runner.RetryAfter(5 * time.Second) - }() - }) + // Run and request a retry + obj.setRetry(true) runner.Run() waitForRunWithRetry("fourth run", t, timer, obj, 5*time.Second) diff --git a/pkg/proxy/winkernel/proxier.go b/pkg/proxy/winkernel/proxier.go index bff18030b32..5ff6307a91b 100644 --- a/pkg/proxy/winkernel/proxier.go +++ b/pkg/proxy/winkernel/proxier.go @@ -753,8 +753,9 @@ func NewProxier( return nil, err } - klog.V(3).InfoS("Record sync param", "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod) - proxier.syncRunner = runner.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod) + klog.V(3).Info("Record sync params", "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod, "maxSyncPeriod", proxyutil.FullSyncPeriod) + proxier.syncRunner = runner.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, proxyutil.FullSyncPeriod) + return proxier, nil } @@ -1192,7 +1193,7 @@ func (proxier *Proxier) handleUpdateLoadbalancerFailure(err error, hnsID, svcIP // This is where all of the hns save/restore calls happen. // assumes proxier.mu is held -func (proxier *Proxier) syncProxyRules() { +func (proxier *Proxier) syncProxyRules() (retryError error) { proxier.mu.Lock() defer proxier.mu.Unlock() @@ -1788,6 +1789,7 @@ func (proxier *Proxier) syncProxyRules() { // This will cleanup stale load balancers which are pending delete // in last iteration proxier.cleanupStaleLoadbalancers() + return } // deleteExistingLoadBalancer checks whether loadbalancer delete is needed or not. From 459188ce25cf4af2f6b9912e6167bc96a1e96b23 Mon Sep 17 00:00:00 2001 From: Antonio Ojea Date: Thu, 1 May 2025 23:28:37 +0000 Subject: [PATCH 5/5] Port BoundedFrequencyRunner from flowcontrol.RateLimiter to clock.Clock Co-authored-by: Dan Winship --- pkg/proxy/runner/bounded_frequency_runner.go | 207 ++----- .../runner/bounded_frequency_runner_test.go | 577 ++++++++++-------- 2 files changed, 365 insertions(+), 419 deletions(-) diff --git a/pkg/proxy/runner/bounded_frequency_runner.go b/pkg/proxy/runner/bounded_frequency_runner.go index 525fa8f3c6d..b282381488d 100644 --- a/pkg/proxy/runner/bounded_frequency_runner.go +++ b/pkg/proxy/runner/bounded_frequency_runner.go @@ -18,12 +18,11 @@ package runner import ( "fmt" - "sync" "time" - "k8s.io/client-go/util/flowcontrol" - + utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/klog/v2" + "k8s.io/utils/clock" ) // BoundedFrequencyRunner manages runs of a user-provided work function. @@ -36,96 +35,14 @@ type BoundedFrequencyRunner struct { run chan struct{} // try an async run - mu sync.Mutex // guards runs of fn and all mutations - fn func() error // the work function - lastRun time.Time // time of last run - timer timer // timer for deferred runs - limiter rateLimiter // rate limiter for on-demand runs + fn func() error // the work function + minIntervalTimer clock.Timer + nextRunTimer clock.Timer // Combined timer for maxInterval and retryInterval logic + clock clock.Clock } -// designed so that flowcontrol.RateLimiter satisfies -type rateLimiter interface { - TryAccept() bool - Stop() -} - -type nullLimiter struct{} - -func (nullLimiter) TryAccept() bool { - return true -} - -func (nullLimiter) Stop() {} - -var _ rateLimiter = nullLimiter{} - -// for testing -type timer interface { - // C returns the timer's selectable channel. - C() <-chan time.Time - - // See time.Timer.Reset. - Reset(d time.Duration) bool - - // See time.Timer.Stop. - Stop() bool - - // See time.Now. - Now() time.Time - - // Remaining returns the time until the timer will go off (if it is running). - Remaining() time.Duration - - // See time.Since. - Since(t time.Time) time.Duration - - // See time.Sleep. - Sleep(d time.Duration) -} - -// implement our timer in terms of std time.Timer. -type realTimer struct { - timer *time.Timer - next time.Time -} - -func (rt *realTimer) C() <-chan time.Time { - return rt.timer.C -} - -func (rt *realTimer) Reset(d time.Duration) bool { - rt.next = time.Now().Add(d) - return rt.timer.Reset(d) -} - -func (rt *realTimer) Stop() bool { - return rt.timer.Stop() -} - -func (rt *realTimer) Now() time.Time { - return time.Now() -} - -func (rt *realTimer) Remaining() time.Duration { - return rt.next.Sub(time.Now()) -} - -func (rt *realTimer) Since(t time.Time) time.Duration { - return time.Since(t) -} - -func (rt *realTimer) Sleep(d time.Duration) { - time.Sleep(d) -} - -var _ timer = &realTimer{} - // NewBoundedFrequencyRunner creates and returns a new BoundedFrequencyRunner. -// This runner manages the execution frequency of the provided work function `fn`. -// -// All runs will be async to the caller of BoundedFrequencyRunner.Run, but -// multiple runs are serialized. If the function needs to hold locks, it must -// take them internally. +// This runner manages the execution frequency of the provided function `fn`. // // The runner guarantees two properties: // 1. Minimum Interval (`minInterval`): At least `minInterval` must pass between @@ -143,13 +60,11 @@ var _ timer = &realTimer{} // (unless another trigger, like `Run()` or `maxInterval`, causes it to run sooner). Any // successful run will abort the retry attempt. func NewBoundedFrequencyRunner(name string, fn func() error, minInterval, retryInterval, maxInterval time.Duration) *BoundedFrequencyRunner { - timer := &realTimer{timer: time.NewTimer(0)} // will tick immediately - <-timer.C() // consume the first tick - return construct(name, fn, minInterval, retryInterval, maxInterval, timer) + return construct(name, fn, minInterval, retryInterval, maxInterval, clock.RealClock{}) } // Make an instance with dependencies injected. -func construct(name string, fn func() error, minInterval, retryInterval, maxInterval time.Duration, timer timer) *BoundedFrequencyRunner { +func construct(name string, fn func() error, minInterval, retryInterval, maxInterval time.Duration, clock clock.Clock) *BoundedFrequencyRunner { if maxInterval < minInterval { panic(fmt.Sprintf("%s: maxInterval (%v) must be >= minInterval (%v)", name, maxInterval, minInterval)) } @@ -163,14 +78,9 @@ func construct(name string, fn func() error, minInterval, retryInterval, maxInte maxInterval: maxInterval, run: make(chan struct{}, 1), - timer: timer, - } - if minInterval == 0 { - bfr.limiter = nullLimiter{} - } else { - qps := float32(time.Second) / float32(minInterval) - bfr.limiter = flowcontrol.NewTokenBucketRateLimiterWithClock(qps, 1, timer) + clock: clock, } + return bfr } @@ -178,17 +88,56 @@ func construct(name string, fn func() error, minInterval, retryInterval, maxInte // called as a goroutine. func (bfr *BoundedFrequencyRunner) Loop(stop <-chan struct{}) { klog.V(3).InfoS("Loop running", "runner", bfr.name) - bfr.timer.Reset(bfr.maxInterval) + defer close(bfr.run) + + bfr.minIntervalTimer = bfr.clock.NewTimer(bfr.minInterval) + defer bfr.minIntervalTimer.Stop() + + // Initialize nextRunTimer with maxInterval + bfr.nextRunTimer = bfr.clock.NewTimer(bfr.maxInterval) + defer bfr.nextRunTimer.Stop() + for { select { case <-stop: - bfr.stop() klog.V(3).InfoS("Loop stopping", "runner", bfr.name) return - case <-bfr.timer.C(): - bfr.tryRun() + case <-bfr.nextRunTimer.C(): // Wait on the single timer case <-bfr.run: - bfr.tryRun() + } + + // stop the timers here to allow the tests using the fake clock to synchronize + // with the fakeClock.HasWaiters() method. The timers are reset after the function + // is executed. + bfr.minIntervalTimer.Stop() + bfr.nextRunTimer.Stop() + + var err error + // avoid crashing if the function executed crashes + func() { + defer utilruntime.HandleCrash() + err = bfr.fn() + }() + + // Determine the next interval based on the result + nextInterval := bfr.maxInterval + if err != nil { + // If error, ensure next run is within retryInterval and maxInterval + if bfr.retryInterval < nextInterval { + nextInterval = bfr.retryInterval + } + klog.V(3).InfoS("scheduling retry", "runner", bfr.name, "interval", nextInterval, "error", err) + } + // Reset the timers + bfr.minIntervalTimer.Reset(bfr.minInterval) + bfr.nextRunTimer.Reset(nextInterval) + + // Wait for minInterval before looping + select { + case <-stop: + klog.V(3).InfoS("Loop stopping", "runner", bfr.name) + return + case <-bfr.minIntervalTimer.C(): } } } @@ -204,49 +153,3 @@ func (bfr *BoundedFrequencyRunner) Run() { default: } } - -// assumes the lock is not held -func (bfr *BoundedFrequencyRunner) stop() { - bfr.mu.Lock() - defer bfr.mu.Unlock() - bfr.limiter.Stop() - bfr.timer.Stop() -} - -// assumes the lock is not held -func (bfr *BoundedFrequencyRunner) tryRun() { - bfr.mu.Lock() - defer bfr.mu.Unlock() - - if bfr.limiter.TryAccept() { - // We're allowed to run the function right now. - err := bfr.fn() - - bfr.lastRun = bfr.timer.Now() - bfr.timer.Stop() - - nextInterval := bfr.maxInterval - if err != nil { - // an error will schedule a retry after the retryInterval, - // any successful run before that will stop the retry attempt. - nextInterval = bfr.retryInterval - klog.V(3).InfoS("scheduling retry", "runner", bfr.name, "interval", nextInterval, "error", err) - } - bfr.timer.Reset(nextInterval) - return - } - - // It can't run right now, figure out when it can run next. - elapsed := bfr.timer.Since(bfr.lastRun) // how long since last run - nextPossible := bfr.minInterval - elapsed // time to next possible run - nextScheduled := bfr.timer.Remaining() // time to next scheduled run - klog.V(4).InfoS("can't run", "runner", bfr.name, "elapsed", elapsed, "nextPossible", nextPossible, "nextScheduled", nextScheduled) - - // It's hard to avoid race conditions in the unit tests unless we always reset - // the timer here, even when it's unchanged - if nextPossible < nextScheduled { - nextScheduled = nextPossible - } - bfr.timer.Stop() - bfr.timer.Reset(nextScheduled) -} diff --git a/pkg/proxy/runner/bounded_frequency_runner_test.go b/pkg/proxy/runner/bounded_frequency_runner_test.go index 5c7ea48d4f0..b8c7953c4ac 100644 --- a/pkg/proxy/runner/bounded_frequency_runner_test.go +++ b/pkg/proxy/runner/bounded_frequency_runner_test.go @@ -19,354 +19,397 @@ package runner import ( "fmt" "sync" + "sync/atomic" "testing" "time" + + clock "k8s.io/utils/clock/testing" ) // Track calls to the managed function. type receiver struct { - lock sync.Mutex - run bool - retry bool + counter atomic.Int32 + // counterCh signals completion of F() and sends the new count. + // It's unbuffered to make the send in F() blocking. + counterCh chan int + resultMu sync.RWMutex + result error } func (r *receiver) F() error { - r.lock.Lock() - defer r.lock.Unlock() - r.run = true - - if r.retry { - r.retry = false - return fmt.Errorf("retry") - } - return nil + newCount := r.counter.Add(1) + // Blocking send: F() will wait here until the test reads from counterCh. + r.counterCh <- int(newCount) + r.resultMu.RLock() + defer r.resultMu.RUnlock() + return r.result } -func (r *receiver) reset() bool { - r.lock.Lock() - defer r.lock.Unlock() - was := r.run - r.run = false - return was -} - -func (r *receiver) setRetry(retry bool) { - r.lock.Lock() - defer r.lock.Unlock() - r.retry = retry -} - -// A single change event in the fake timer. -type timerUpdate struct { - active bool - next time.Duration // iff active == true -} - -// Fake time. -type fakeTimer struct { - c chan time.Time - - lock sync.Mutex - now time.Time - timeout time.Time - active bool - - updated chan timerUpdate -} - -func newFakeTimer() *fakeTimer { - ft := &fakeTimer{ - now: time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC), - c: make(chan time.Time), - updated: make(chan timerUpdate), - } - return ft -} - -func (ft *fakeTimer) C() <-chan time.Time { - return ft.c -} - -func (ft *fakeTimer) Reset(in time.Duration) bool { - ft.lock.Lock() - defer ft.lock.Unlock() - - was := ft.active - ft.active = true - ft.timeout = ft.now.Add(in) - ft.updated <- timerUpdate{ - active: true, - next: in, - } - return was -} - -func (ft *fakeTimer) Stop() bool { - ft.lock.Lock() - defer ft.lock.Unlock() - - was := ft.active - ft.active = false - ft.updated <- timerUpdate{ - active: false, - } - return was -} - -func (ft *fakeTimer) Now() time.Time { - ft.lock.Lock() - defer ft.lock.Unlock() - - return ft.now -} - -func (ft *fakeTimer) Remaining() time.Duration { - ft.lock.Lock() - defer ft.lock.Unlock() - - return ft.timeout.Sub(ft.now) -} - -func (ft *fakeTimer) Since(t time.Time) time.Duration { - ft.lock.Lock() - defer ft.lock.Unlock() - - return ft.now.Sub(t) -} - -func (ft *fakeTimer) Sleep(d time.Duration) { - // ft.advance grabs ft.lock - ft.advance(d) -} - -// advance the current time. -func (ft *fakeTimer) advance(d time.Duration) { - ft.lock.Lock() - defer ft.lock.Unlock() - - ft.now = ft.now.Add(d) - if ft.active && !ft.now.Before(ft.timeout) { - ft.active = false - ft.c <- ft.timeout +func newReceiver() *receiver { + return &receiver{ + counterCh: make(chan int), } } -// return the calling line number (for printing) -// test the timer's state -func checkTimer(name string, t *testing.T, upd timerUpdate, active bool, next time.Duration) { - t.Helper() - if upd.active != active { - t.Fatalf("%s: expected timer active=%v", name, active) - } - if active && upd.next != next { - t.Fatalf("%s: expected timer to be %v, got %v", name, next, upd.next) - } +func (r *receiver) calls() <-chan int { + return r.counterCh } -// test and reset the receiver's state -func checkReceiver(name string, t *testing.T, receiver *receiver, expected bool) { - t.Helper() - triggered := receiver.reset() - if expected && !triggered { - t.Fatalf("%s: function should have been called", name) - } else if !expected && triggered { - t.Fatalf("%s: function should not have been called", name) - } +func (r *receiver) setReturnValue(err error) { + r.resultMu.Lock() + defer r.resultMu.Unlock() + r.result = err } -// Durations embedded in test cases depend on these. -var minInterval = 1 * time.Second -var retryInterval = 5 * time.Second -var maxInterval = 10 * time.Second - -func waitForReset(name string, t *testing.T, timer *fakeTimer, obj *receiver, expectCall bool, expectNext time.Duration) { - t.Helper() - upd := <-timer.updated // wait for stop - checkReceiver(name, t, obj, expectCall) - checkReceiver(name, t, obj, false) // prove post-condition - checkTimer(name, t, upd, false, 0) - upd = <-timer.updated // wait for reset - checkTimer(name, t, upd, true, expectNext) -} - -func waitForRun(name string, t *testing.T, timer *fakeTimer, obj *receiver) { - t.Helper() - waitForReset(name, t, timer, obj, true, maxInterval) -} - -func waitForRunWithRetry(name string, t *testing.T, timer *fakeTimer, obj *receiver, expectNext time.Duration) { - t.Helper() - waitForReset(name, t, timer, obj, true, expectNext) -} - -func waitForDefer(name string, t *testing.T, timer *fakeTimer, obj *receiver, expectNext time.Duration) { - t.Helper() - waitForReset(name, t, timer, obj, false, expectNext) -} - -func waitForNothing(name string, t *testing.T, timer *fakeTimer, obj *receiver) { +// assertCalls waits for the receiver's function to be called and asserts that +// the total call count matches expectedCalls. It fails the test if the timeout is reached +// or if the call count doesn't match. +func assertCalls(t *testing.T, r *receiver, expectedCalls int) { t.Helper() select { - case <-timer.c: - t.Fatalf("%s: unexpected timer tick", name) - case upd := <-timer.updated: - t.Fatalf("%s: unexpected timer update %v", name, upd) - default: + case calls := <-r.calls(): + if calls != expectedCalls { + t.Fatalf("expected %d calls, but got %d", expectedCalls, calls) + } + case <-time.After(1 * time.Second): + t.Fatalf("timed out waiting for function execution (expected %d calls, got %d)", expectedCalls, r.counter.Load()) + } +} + +// assertNoCalls waits for 100 millisecond and asserts that the receiver's +// function was *not* called during that time. It fails the test if a call is detected. +func assertNoCalls(t *testing.T, r *receiver) { + t.Helper() + select { + case calls := <-r.calls(): + t.Fatalf("unexpected function execution detected (call count: %d)", calls) + case <-time.After(100 * time.Millisecond): } - checkReceiver(name, t, obj, false) } func Test_BoundedFrequencyRunner(t *testing.T) { - obj := &receiver{} - timer := newFakeTimer() - runner := construct("test-runner", obj.F, minInterval, retryInterval, maxInterval, timer) + var minInterval = 1 * time.Second + var retryInterval = 5 * time.Second + var maxInterval = 10 * time.Second + obj := newReceiver() + fakeClock := clock.NewFakeClock(time.Now()) + runner := construct("test-runner", obj.F, minInterval, retryInterval, maxInterval, fakeClock) stop := make(chan struct{}) + defer close(stop) - var upd timerUpdate - - // Start. go runner.Loop(stop) - upd = <-timer.updated // wait for initial time to be set to max - checkTimer("init", t, upd, true, maxInterval) - checkReceiver("init", t, obj, false) // Run once, immediately. // rel=0ms runner.Run() - waitForRun("first run", t, timer, obj) - - // Run again, before minInterval expires. - timer.advance(500 * time.Millisecond) // rel=500ms + assertCalls(t, obj, 1) + // wait for the timers to be reset + for fakeClock.Waiters() != 2 { + time.Sleep(1 * time.Millisecond) + } + // Run again, before minInterval expires. No execution expected. + fakeClock.Step(500 * time.Millisecond) // rel=500ms runner.Run() - waitForDefer("too soon after first", t, timer, obj, 500*time.Millisecond) + assertNoCalls(t, obj) - // Run again, before minInterval expires. - timer.advance(499 * time.Millisecond) // rel=999ms + // Run again, before minInterval expires. No execution expected. + fakeClock.Step(499 * time.Millisecond) // rel=999ms runner.Run() - waitForDefer("still too soon after first", t, timer, obj, 1*time.Millisecond) + assertNoCalls(t, obj) // Do the deferred run - timer.advance(1 * time.Millisecond) // rel=1000ms - waitForRun("second run", t, timer, obj) + fakeClock.Step(1 * time.Millisecond) // rel=1000ms + assertCalls(t, obj, 2) - // Try again immediately runner.Run() - waitForDefer("too soon after second", t, timer, obj, 1*time.Second) + assertNoCalls(t, obj) - // Run again, before minInterval expires. - timer.advance(1 * time.Millisecond) // rel=1ms + // Run again, before minInterval expires. No execution expected. + fakeClock.Step(1 * time.Millisecond) // rel=1ms runner.Run() - waitForDefer("still too soon after second", t, timer, obj, 999*time.Millisecond) + assertNoCalls(t, obj) - // Ensure that we don't run again early - timer.advance(998 * time.Millisecond) // rel=999ms - waitForNothing("premature", t, timer, obj) + // Ensure that we don't run again early. No execution expected. + fakeClock.Step(998 * time.Millisecond) // rel=999ms + assertNoCalls(t, obj) // Do the deferred run - timer.advance(1 * time.Millisecond) // rel=1000ms - waitForRun("third run", t, timer, obj) - - // Let minInterval pass, but there are no runs queued - timer.advance(1 * time.Second) // rel=1000ms - waitForNothing("minInterval", t, timer, obj) + fakeClock.Step(1 * time.Millisecond) // rel=1000ms + assertCalls(t, obj, 3) + // wait for the timers to be reset + for fakeClock.Waiters() != 2 { + time.Sleep(1 * time.Millisecond) + } + // Let minInterval pass, but there are no runs queued. No execution expected. + fakeClock.Step(1 * time.Second) // rel=1000ms + assertNoCalls(t, obj) // Let maxInterval pass - timer.advance(9 * time.Second) // rel=10000ms - waitForRun("maxInterval", t, timer, obj) - - // Run again, before minInterval expires. - timer.advance(1 * time.Millisecond) // rel=1ms + fakeClock.Step(maxInterval) // rel=10000ms + assertCalls(t, obj, 4) + // wait for the timers to be reset + for fakeClock.Waiters() != 2 { + time.Sleep(1 * time.Millisecond) + } + // Run again, before minInterval expires. No execution expected. + fakeClock.Step(1 * time.Millisecond) // rel=1ms runner.Run() - waitForDefer("too soon after maxInterval run", t, timer, obj, 999*time.Millisecond) + assertNoCalls(t, obj) // Let minInterval pass - timer.advance(999 * time.Millisecond) // rel=1000ms - waitForRun("fifth run", t, timer, obj) - - // Clean up. - stop <- struct{}{} - // a message is sent to time.updated in func Stop() at the end of the child goroutine - // to terminate the child, a receive on time.updated is needed here - <-timer.updated + fakeClock.Step(999 * time.Millisecond) // rel=1000ms + assertCalls(t, obj, 5) } func Test_BoundedFrequencyRunnerRetry(t *testing.T) { - obj := &receiver{} - timer := newFakeTimer() - runner := construct("test-runner", obj.F, minInterval, retryInterval, maxInterval, timer) + var minInterval = 1 * time.Second + var retryInterval = 5 * time.Second + var maxInterval = 10 * time.Second + obj := newReceiver() + fakeClock := clock.NewFakeClock(time.Now()) + runner := construct("test-runner", obj.F, minInterval, retryInterval, maxInterval, fakeClock) stop := make(chan struct{}) + defer close(stop) - var upd timerUpdate - - // Start. go runner.Loop(stop) - upd = <-timer.updated // wait for initial time to be set to max - checkTimer("init", t, upd, true, maxInterval) - checkReceiver("init", t, obj, false) // Run once, immediately, and queue a retry // rel=0ms - obj.setRetry(true) + obj.setReturnValue(fmt.Errorf("sync error")) runner.Run() - waitForRunWithRetry("first run", t, timer, obj, 5*time.Second) + assertCalls(t, obj, 1) + // wait for the timers to be reset + for fakeClock.Waiters() != 2 { + time.Sleep(1 * time.Millisecond) + } + + // next run will succeed + obj.setReturnValue(nil) + assertNoCalls(t, obj) // Nothing happens... - timer.advance(time.Second) // rel=1000ms - waitForNothing("minInterval, nothing queued", t, timer, obj) + fakeClock.Step(minInterval) // rel=1000ms + assertNoCalls(t, obj) // After retryInterval, function is called - timer.advance(4 * time.Second) // rel=5000ms - waitForRun("retry", t, timer, obj) + fakeClock.Step(4 * time.Second) // rel=5000ms + assertCalls(t, obj, 2) + // wait for the timers to be reset + for fakeClock.Waiters() != 2 { + time.Sleep(1 * time.Millisecond) + } - // Run again, before minInterval expires. - timer.advance(499 * time.Millisecond) // rel=499ms + // Run again, before minInterval expires and trigger a retry + fakeClock.Step(499 * time.Millisecond) // rel=499ms + obj.setReturnValue(fmt.Errorf("sync error")) runner.Run() - waitForDefer("too soon after retry", t, timer, obj, 501*time.Millisecond) + assertNoCalls(t, obj) - // Do the deferred run, have it queue another retry - obj.setRetry(true) - timer.advance(501 * time.Millisecond) // rel=1000ms - waitForRunWithRetry("second run", t, timer, obj, 5*time.Second) + // Do the deferred run, queue another retry after it returns + fakeClock.Step(501 * time.Millisecond) // rel=1000ms + assertCalls(t, obj, 3) + + // next run will succeed + obj.setReturnValue(nil) + assertNoCalls(t, obj) // Wait for minInterval to pass - timer.advance(time.Second) // rel=1000ms - waitForNothing("minInterval, nothing queued", t, timer, obj) + fakeClock.Step(time.Second) // rel=1000ms + assertNoCalls(t, obj) - // Now do another run + // Now do another successful that abort the retry runner.Run() - waitForRun("third run", t, timer, obj) + assertCalls(t, obj, 4) + // wait for the timers to be reset + for fakeClock.Waiters() != 2 { + time.Sleep(1 * time.Millisecond) + } // Retry was cancelled because we already ran - timer.advance(4 * time.Second) - waitForNothing("retry cancelled", t, timer, obj) + fakeClock.Step(4 * time.Second) + assertNoCalls(t, obj) - // Run and request a retry - obj.setRetry(true) + // New run will trigger a retry. + obj.setReturnValue(fmt.Errorf("sync error")) runner.Run() - waitForRunWithRetry("fourth run", t, timer, obj, 5*time.Second) + assertCalls(t, obj, 5) + for fakeClock.Waiters() != 2 { // wait for retryIntervalTimer + time.Sleep(1 * time.Millisecond) + } + + // next run will succeed + obj.setReturnValue(nil) + assertNoCalls(t, obj) // Call Run again before minInterval passes - timer.advance(100 * time.Millisecond) // rel=100ms + fakeClock.Step(100 * time.Millisecond) // rel=100ms runner.Run() - waitForDefer("too soon after fourth run", t, timer, obj, 900*time.Millisecond) + assertNoCalls(t, obj) // Deferred run will run after minInterval passes - timer.advance(900 * time.Millisecond) // rel=1000ms - waitForRun("fifth run", t, timer, obj) + fakeClock.Step(900 * time.Millisecond) // rel=1000ms + assertCalls(t, obj, 6) + // wait for the timers to be reset + for fakeClock.Waiters() != 2 { + time.Sleep(1 * time.Millisecond) + } // Retry was cancelled because we already ran - timer.advance(4 * time.Second) // rel=4s since run, 5s since RetryAfter - waitForNothing("retry cancelled", t, timer, obj) + fakeClock.Step(4 * time.Second) // rel=4s since run, 5s since RetryAfter + assertNoCalls(t, obj) // Rerun happens after maxInterval - timer.advance(5 * time.Second) // rel=9s since run, 10s since RetryAfter - waitForNothing("premature", t, timer, obj) - timer.advance(time.Second) // rel=10s since run - waitForRun("maxInterval", t, timer, obj) + fakeClock.Step(5 * time.Second) // rel=9s since run, 10s since RetryAfter + assertNoCalls(t, obj) - // Clean up. - stop <- struct{}{} - // a message is sent to time.updated in func Stop() at the end of the child goroutine - // to terminate the child, a receive on time.updated is needed here - <-timer.updated + fakeClock.Step(time.Second) // rel=10s since run + assertCalls(t, obj, 7) +} + +func Test_BoundedFrequencyRunnerRetryShorterThanMinInterval(t *testing.T) { + var minInterval = 5 * time.Second + var retryInterval = 1 * time.Second // Shorter than minInterval + var maxInterval = 10 * time.Second + obj := newReceiver() + fakeClock := clock.NewFakeClock(time.Now()) + runner := construct("test-runner-short-retry", obj.F, minInterval, retryInterval, maxInterval, fakeClock) + stop := make(chan struct{}) + defer close(stop) + + go runner.Loop(stop) + + // Run once immediately and trigger a retry. + // rel=0s + obj.setReturnValue(fmt.Errorf("sync error")) + runner.Run() + assertCalls(t, obj, 1) + for fakeClock.Waiters() != 2 { + time.Sleep(1 * time.Millisecond) + } + + // next run will succeed + obj.setReturnValue(nil) + assertNoCalls(t, obj) + + // Advance clock past retryInterval, but still within minInterval. + // rel=1s + fakeClock.Step(retryInterval) + assertNoCalls(t, obj) // Still shouldn't run because minInterval hasn't passed since run 1 finished. + + // Advance clock just before minInterval expires. + // rel=4.999s + fakeClock.Step(minInterval - retryInterval - 1*time.Millisecond) + assertNoCalls(t, obj) + + // Advance clock past minInterval. The retry should now trigger the run. + // rel=5s + fakeClock.Step(1 * time.Millisecond) + assertCalls(t, obj, 2) // Run happens now, triggered by the earlier retry, respecting minInterval. + // wait for the timers to be reset + for fakeClock.Waiters() != 2 { + time.Sleep(1 * time.Millisecond) + } + // Let maxInterval pass without any Run() or Retry() calls. + fakeClock.Step(maxInterval) // rel=10s since run 2 + assertCalls(t, obj, 3) +} + +func TestBoundedFrequencyRunner_Run_RunsAgainAfterMinInterval_RealClock(t *testing.T) { + // Use relatively short intervals for real clock testing + var minInterval = 500 * time.Millisecond + var retryInterval = 800 * time.Millisecond + var maxInterval = 1500 * time.Millisecond + obj := newReceiver() + runner := NewBoundedFrequencyRunner("test-runner", obj.F, minInterval, retryInterval, maxInterval) + + stopCh := make(chan struct{}) + defer close(stopCh) + go runner.Loop(stopCh) + + runner.Run() // First run + assertCalls(t, obj, 1) + + time.Sleep(2 * minInterval) + assertNoCalls(t, obj) + + runner.Run() // Second run + assertCalls(t, obj, 2) +} + +func TestBoundedFrequencyRunner_Run_DoesNotRunBeforeMinInterval_RealClock(t *testing.T) { + // Use relatively short intervals for real clock testing + var minInterval = 500 * time.Millisecond + var retryInterval = 800 * time.Millisecond + var maxInterval = 1500 * time.Millisecond + obj := newReceiver() + runner := NewBoundedFrequencyRunner("test-runner", obj.F, minInterval, retryInterval, maxInterval) + + stopCh := make(chan struct{}) + defer close(stopCh) + go runner.Loop(stopCh) + + runner.Run() // First run + assertCalls(t, obj, 1) + + time.Sleep(minInterval / 4) + runner.Run() + assertNoCalls(t, obj) +} + +func TestBoundedFrequencyRunner_RunAfterMaxInterval_RealClock(t *testing.T) { + // Use relatively short intervals for real clock testing + var minInterval = 100 * time.Millisecond + var retryInterval = 200 * time.Millisecond + var maxInterval = 500 * time.Millisecond + obj := newReceiver() + runner := NewBoundedFrequencyRunner("test-runner", obj.F, minInterval, retryInterval, maxInterval) + + stopCh := make(chan struct{}) + defer close(stopCh) + go runner.Loop(stopCh) + + assertNoCalls(t, obj) + + time.Sleep(maxInterval) + assertCalls(t, obj, 1) +} + +func Test_BoundedFrequencyRunnerRetry_RealClock(t *testing.T) { + // Use relatively short intervals for real clock testing + var minInterval = 100 * time.Millisecond + var retryInterval = 500 * time.Millisecond + var maxInterval = 10 * time.Second + + obj := newReceiver() + // Use the real clock constructor + runner := NewBoundedFrequencyRunner("test-runner-real-clock", obj.F, minInterval, retryInterval, maxInterval) + + stopCh := make(chan struct{}) + defer close(stopCh) + go runner.Loop(stopCh) + + t.Log("Triggering first retry") + // Run once immediately and trigger a retry. + // rel=0s + obj.setReturnValue(fmt.Errorf("sync error")) + runner.Run() + assertCalls(t, obj, 1) + + // Check before retryInterval + time.Sleep(retryInterval / 4) + assertNoCalls(t, obj) + + // Check after retryInterval + time.Sleep(retryInterval) // Wait past retryInterval + assertCalls(t, obj, 2) + + // Check after retryInterval (relative to the *first* Retry call in this batch) + time.Sleep(retryInterval) + assertCalls(t, obj, 3) + + time.Sleep(retryInterval / 8) + assertNoCalls(t, obj) + + time.Sleep(retryInterval) // Wait past the new retryInterval + assertCalls(t, obj, 4) }