Merge pull request #131615 from danwinship/proxy-bfr

update BoundedFrequencyRunner for kube-proxy
2026-08-08 15:10:43 +00:00 · 2025-07-01 09:21:24 -07:00
parent b99ca3f736 459188ce25
commit c3b06a5366
9 changed files with 605 additions and 32 deletions
--- 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) {
--- 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
 }

--- 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
--- 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
 }

--- 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.
--- 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
 }
--- a/pkg/proxy/runner/bounded_frequency_runner.go
+++ 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:
+	}
+}
--- a/pkg/proxy/runner/bounded_frequency_runner_test.go
+++ 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)
+}
--- 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.