diff --git a/pkg/proxy/iptables/proxier.go b/pkg/proxy/iptables/proxier.go index 09d5abefce8..4d1b257b24c 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 @@ -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, "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 = async.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, proxyutil.FullSyncPeriod, burstSyncs) + 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) @@ -796,7 +795,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() @@ -830,7 +829,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() } @@ -1599,6 +1598,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 1132203fee8..5629e3129b9 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, 30*time.Second, time.Minute) return p } diff --git a/pkg/proxy/ipvs/proxier.go b/pkg/proxy/ipvs/proxier.go index ccdb42d0308..82846e3c1f4 100644 --- a/pkg/proxy/ipvs/proxier.go +++ b/pkg/proxy/ipvs/proxier.go @@ -50,8 +50,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" @@ -188,7 +188,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 @@ -401,9 +401,10 @@ 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 = async.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, burstSyncs) + + 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 } @@ -920,7 +921,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() @@ -1515,6 +1516,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 900a0c6570c..5400b49ac54 100644 --- a/pkg/proxy/ipvs/proxier_test.go +++ b/pkg/proxy/ipvs/proxier_test.go @@ -54,9 +54,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" @@ -168,7 +168,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, 30*time.Second, time.Minute) return p } diff --git a/pkg/proxy/nftables/proxier.go b/pkg/proxy/nftables/proxier.go index 14f49c893d4..b2daee94616 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 @@ -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, "maxSyncPeriod", proxyutil.FullSyncPeriod) // 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, syncPeriod, proxyutil.FullSyncPeriod) return proxier, nil } @@ -1163,7 +1162,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() @@ -1202,7 +1201,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. @@ -1888,6 +1887,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 52207148544..675ade0c194 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, 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 new file mode 100644 index 00000000000..b282381488d --- /dev/null +++ b/pkg/proxy/runner/bounded_frequency_runner.go @@ -0,0 +1,155 @@ +/* +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" + "time" + + utilruntime "k8s.io/apimachinery/pkg/util/runtime" + "k8s.io/klog/v2" + "k8s.io/utils/clock" +) + +// 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 + 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 + + fn func() error // the work function + minIntervalTimer clock.Timer + nextRunTimer clock.Timer // Combined timer for maxInterval and retryInterval logic + clock clock.Clock +} + +// NewBoundedFrequencyRunner creates and returns a new BoundedFrequencyRunner. +// 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 +// 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). +// +// `maxInterval` must be greater than or equal to `minInterval`; otherwise, +// this function will panic. +// +// 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 { + 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, clock clock.Clock) *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, + retryInterval: retryInterval, + maxInterval: maxInterval, + + run: make(chan struct{}, 1), + clock: clock, + } + + 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).InfoS("Loop running", "runner", bfr.name) + 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: + klog.V(3).InfoS("Loop stopping", "runner", bfr.name) + return + case <-bfr.nextRunTimer.C(): // Wait on the single timer + case <-bfr.run: + } + + // 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(): + } + } +} + +// Run the work function as soon as possible. If this is called while Loop is not +// running, the call may be deferred indefinitely. +// 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 bfr.run is empty, push an element onto it. Otherwise, do nothing. + select { + case bfr.run <- struct{}{}: + default: + } +} 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..b8c7953c4ac --- /dev/null +++ b/pkg/proxy/runner/bounded_frequency_runner_test.go @@ -0,0 +1,415 @@ +/* +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" + "sync/atomic" + "testing" + "time" + + clock "k8s.io/utils/clock/testing" +) + +// Track calls to the managed function. +type receiver struct { + 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 { + 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 newReceiver() *receiver { + return &receiver{ + counterCh: make(chan int), + } +} + +func (r *receiver) calls() <-chan int { + return r.counterCh +} + +func (r *receiver) setReturnValue(err error) { + r.resultMu.Lock() + defer r.resultMu.Unlock() + r.result = err +} + +// 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 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): + } +} + +func Test_BoundedFrequencyRunner(t *testing.T) { + 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) + + go runner.Loop(stop) + + // Run once, immediately. + // rel=0ms + runner.Run() + 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() + assertNoCalls(t, obj) + + // Run again, before minInterval expires. No execution expected. + fakeClock.Step(499 * time.Millisecond) // rel=999ms + runner.Run() + assertNoCalls(t, obj) + + // Do the deferred run + fakeClock.Step(1 * time.Millisecond) // rel=1000ms + assertCalls(t, obj, 2) + + runner.Run() + assertNoCalls(t, obj) + + // Run again, before minInterval expires. No execution expected. + fakeClock.Step(1 * time.Millisecond) // rel=1ms + runner.Run() + assertNoCalls(t, 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 + 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 + 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() + assertNoCalls(t, obj) + + // Let minInterval pass + fakeClock.Step(999 * time.Millisecond) // rel=1000ms + assertCalls(t, obj, 5) +} + +func Test_BoundedFrequencyRunnerRetry(t *testing.T) { + 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) + + go runner.Loop(stop) + + // Run once, immediately, and queue a retry + // rel=0ms + obj.setReturnValue(fmt.Errorf("sync error")) + runner.Run() + 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... + fakeClock.Step(minInterval) // rel=1000ms + assertNoCalls(t, obj) + + // After retryInterval, function is called + 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 and trigger a retry + fakeClock.Step(499 * time.Millisecond) // rel=499ms + obj.setReturnValue(fmt.Errorf("sync error")) + runner.Run() + assertNoCalls(t, obj) + + // 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 + fakeClock.Step(time.Second) // rel=1000ms + assertNoCalls(t, obj) + + // Now do another successful that abort the retry + runner.Run() + 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 + fakeClock.Step(4 * time.Second) + assertNoCalls(t, obj) + + // New run will trigger a retry. + obj.setReturnValue(fmt.Errorf("sync error")) + runner.Run() + 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 + fakeClock.Step(100 * time.Millisecond) // rel=100ms + runner.Run() + assertNoCalls(t, obj) + + // Deferred run will run after minInterval passes + 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 + fakeClock.Step(4 * time.Second) // rel=4s since run, 5s since RetryAfter + assertNoCalls(t, obj) + + // Rerun happens after maxInterval + fakeClock.Step(5 * time.Second) // rel=9s since run, 10s since RetryAfter + assertNoCalls(t, obj) + + 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) +} diff --git a/pkg/proxy/winkernel/proxier.go b/pkg/proxy/winkernel/proxier.go index 91fb6fad8d4..5ff6307a91b 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 @@ -753,9 +753,9 @@ func NewProxier( return nil, err } - 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) + 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 } @@ -1193,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() @@ -1789,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.