Skip to content

Commit 3d066d8

Browse files
michaelawyumichaelawyu
authored
feat: improve the default setup for hub agent leader election to allow better scalability/stability (kubefleet-dev#414)
1 parent e68d911 commit 3d066d8

5 files changed

Lines changed: 208 additions & 20 deletions

File tree

‎cmd/hubagent/main.go‎

Lines changed: 33 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ import (
2828
"k8s.io/apimachinery/pkg/runtime"
2929
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
3030
clientgoscheme "k8s.io/client-go/kubernetes/scheme"
31+
"k8s.io/client-go/rest"
3132
"k8s.io/klog/v2"
3233
clusterinventory "sigs.k8s.io/cluster-inventory-api/apis/v1alpha1"
3334
ctrl "sigs.k8s.io/controller-runtime"
@@ -101,21 +102,46 @@ func main() {
101102
// Set up controller-runtime logger
102103
ctrl.SetLogger(zap.New(zap.UseDevMode(true)))
103104

104-
config := ctrl.GetConfigOrDie()
105-
config.QPS, config.Burst = float32(opts.CtrlMgrOpts.HubQPS), opts.CtrlMgrOpts.HubBurst
105+
// Create separate configs for the general access purpose and the leader election purpose.
106+
//
107+
// This aims to improve the availability of the hub agent; originally all access to the API
108+
// server would share the same rate limiting configuration, and under adverse conditions (e.g.,
109+
// large volume of concurrent placements) controllers would exhause all tokens in the rate limiter,
110+
// and effectively starve the leader election process (the runtime can no longer renew leases),
111+
// which would trigger the hub agent to restart even though the system remains functional.
112+
defaultCfg := ctrl.GetConfigOrDie()
113+
leaderElectionCfg := rest.CopyConfig(defaultCfg)
114+
115+
defaultCfg.QPS, defaultCfg.Burst = float32(opts.CtrlMgrOpts.HubQPS), opts.CtrlMgrOpts.HubBurst
116+
leaderElectionCfg.QPS, leaderElectionCfg.Burst = float32(opts.LeaderElectionOpts.LeaderElectionQPS), opts.LeaderElectionOpts.LeaderElectionBurst
106117

107118
mgrOpts := ctrl.Options{
108119
Scheme: scheme,
109120
Cache: cache.Options{
110121
SyncPeriod: &opts.CtrlMgrOpts.ResyncPeriod.Duration,
111122
DefaultTransform: cache.TransformStripManagedFields(),
112123
},
113-
LeaderElection: opts.LeaderElectionOpts.LeaderElect,
114-
LeaderElectionID: "136224848560.hub.fleet.azure.com",
115-
LeaderElectionNamespace: opts.LeaderElectionOpts.ResourceNamespace,
124+
LeaderElection: opts.LeaderElectionOpts.LeaderElect,
125+
LeaderElectionConfig: leaderElectionCfg,
126+
// If leader election is enabled, the hub agent by default uses a setup
127+
// with a lease duration of 60 secs, a renew deadline of 45 secs, and a retry period of 5 secs.
128+
// This setup gives the hub agent up to 9 attempts/45 seconds to renew its leadership lease
129+
// before it loses the leadership and restarts.
130+
//
131+
// These values are set significantly higher than the controller-runtime defaults
132+
// (15 seconds, 10 seconds, and 2 seconds respectively), as under heavy loads the hub agent
133+
// might have difficulty renewing its lease in time due to API server side latencies, which
134+
// might further lead to unexpected leadership losses (even when it is the only candidate
135+
// running) and restarts.
136+
//
137+
// Note (chenyu1): a minor side effect with the higher values is that when the agent does restart,
138+
// (or in the future when we do run multiple hub agent replicas), the new leader might have to wait a bit
139+
// longer (up to 60 seconds) to acquire the leadership, which should still be acceptable in most scenarios.
116140
LeaseDuration: &opts.LeaderElectionOpts.LeaseDuration.Duration,
117141
RenewDeadline: &opts.LeaderElectionOpts.RenewDeadline.Duration,
118142
RetryPeriod: &opts.LeaderElectionOpts.RetryPeriod.Duration,
143+
LeaderElectionID: "136224848560.hub.fleet.azure.com",
144+
LeaderElectionNamespace: opts.LeaderElectionOpts.ResourceNamespace,
119145
HealthProbeBindAddress: opts.CtrlMgrOpts.HealthProbeBindAddress,
120146
Metrics: metricsserver.Options{
121147
BindAddress: opts.CtrlMgrOpts.MetricsBindAddress,
@@ -128,7 +154,7 @@ func main() {
128154
if opts.CtrlMgrOpts.EnablePprof {
129155
mgrOpts.PprofBindAddress = fmt.Sprintf(":%d", opts.CtrlMgrOpts.PprofPort)
130156
}
131-
mgr, err := ctrl.NewManager(config, mgrOpts)
157+
mgr, err := ctrl.NewManager(defaultCfg, mgrOpts)
132158
if err != nil {
133159
klog.ErrorS(err, "unable to start controller manager.")
134160
exitWithErrorFunc()
@@ -186,7 +212,7 @@ func main() {
186212
}
187213

188214
ctx := ctrl.SetupSignalHandler()
189-
if err := workload.SetupControllers(ctx, &wg, mgr, config, opts); err != nil {
215+
if err := workload.SetupControllers(ctx, &wg, mgr, defaultCfg, opts); err != nil {
190216
klog.ErrorS(err, "unable to set up controllers")
191217
exitWithErrorFunc()
192218
}

‎cmd/hubagent/options/leaderelection.go‎

Lines changed: 87 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,8 @@ package options
1818

1919
import (
2020
"flag"
21+
"fmt"
22+
"strconv"
2123
"time"
2224

2325
"github.com/kubefleet-dev/kubefleet/pkg/utils"
@@ -52,6 +54,18 @@ type LeaderElectionOptions struct {
5254
// The namespace of the resource object that will be used to lock during leader election cycles.
5355
// This option only applies if leader election is enabled.
5456
ResourceNamespace string
57+
58+
// The QPS limit set to the rate limiter of the Kubernetes client in use by the controller manager
59+
// for leader election purposes. This sets up a separate client-side throttling mechanism specifically
60+
// for lease related operations, mostly to avoid an adverse situation where normal operations
61+
// in the controller manager starve the lease related operations, and thus cause unexpected leadership losses.
62+
LeaderElectionQPS float64
63+
64+
// The burst limit set to the rate limiter of the Kubernetes client in use by the controller manager
65+
// for leader election purposes. This sets up a separate client-side throttling mechanism specifically
66+
// for lease related operations, mostly to avoid an adverse situation where normal operations
67+
// in the controller manager starve the lease related operations, and thus cause unexpected leadership losses.
68+
LeaderElectionBurst int
5569
}
5670

5771
// AddFlags adds flags for LeaderElectionOptions to the specified FlagSet.
@@ -68,23 +82,23 @@ func (o *LeaderElectionOptions) AddFlags(flags *flag.FlagSet) {
6882
flags.DurationVar(
6983
&o.LeaseDuration.Duration,
7084
"leader-lease-duration",
71-
15*time.Second,
85+
60*time.Second,
7286
"The duration of a leader election lease. This is the period where a non-leader candidate will wait after observing a leadership renewal before attempting to acquire leadership of the current leader. And it is also effectively the maximum duration that a leader can be stopped before it is replaced by another candidate. The option only applies if leader election is enabled.",
7387
)
7488

7589
// This input is sent to the controller manager for validation; no further check here.
7690
flags.DurationVar(
7791
&o.RenewDeadline.Duration,
7892
"leader-renew-deadline",
79-
10*time.Second,
93+
45*time.Second,
8094
"The interval between attempts by the acting master to renew a leadership slot before it stops leading. This must be less than or equal to the lease duration. The option only applies if leader election is enabled",
8195
)
8296

8397
// This input is sent to the controller manager for validation; no further check here.
8498
flags.DurationVar(
8599
&o.RetryPeriod.Duration,
86100
"leader-retry-period",
87-
2*time.Second,
101+
5*time.Second,
88102
"The duration the clients should wait between attempting acquisition and renewal of a leadership. The option only applies if leader election is enabled",
89103
)
90104

@@ -95,4 +109,74 @@ func (o *LeaderElectionOptions) AddFlags(flags *flag.FlagSet) {
95109
utils.FleetSystemNamespace,
96110
"The namespace of the resource object that will be used to lock during leader election cycles. The option only applies if leader election is enabled.",
97111
)
112+
113+
flags.Var(
114+
newLeaderElectionQPSValueWithValidation(250, &o.LeaderElectionQPS),
115+
"leader-election-qps",
116+
"The QPS limit set to the rate limiter of the Kubernetes client in use by the controller manager for leader election purposes. This sets up a separate client-side throttling mechanism specifically for lease related operations, mostly to avoid an adverse situation where normal operations in the controller manager starve the lease related operations, and thus cause unexpected leadership losses. Defaults to 250. Use a positive float64 value in the range [10.0, 1000.0], or set a less or equal to zero value to disable client-side throttling.")
117+
118+
flags.Var(
119+
newLeaderElectionBurstValueWithValidation(1000, &o.LeaderElectionBurst),
120+
"leader-election-burst",
121+
"The burst limit set to the rate limiter of the Kubernetes client in use by the controller manager for leader election purposes. This sets up a separate client-side throttling mechanism specifically for lease related operations, mostly to avoid an adverse situation where normal operations in the controller manager starve the lease related operations, and thus cause unexpected leadership losses. Defaults to 1000. Use a positive int value in the range [10, 2000].")
122+
}
123+
124+
// A list of flag variables that allow pluggable validation logic when parsing the input args.
125+
126+
type LeaderElectionQPSValueWithValidation float64
127+
128+
func (v *LeaderElectionQPSValueWithValidation) String() string {
129+
return fmt.Sprintf("%f", *v)
130+
}
131+
132+
func (v *LeaderElectionQPSValueWithValidation) Set(s string) error {
133+
// Some validation is also performed on the controller manager side and the client-go side. Just
134+
// to be on the safer side we also impose some limits here.
135+
qps, err := strconv.ParseFloat(s, 64)
136+
if err != nil {
137+
return fmt.Errorf("failed to parse float64 value: %w", err)
138+
}
139+
140+
if qps <= 0.0 {
141+
// Disable client-side throttling.
142+
*v = -1.0
143+
return nil
144+
}
145+
146+
if qps < 10.0 || qps > 1000.0 {
147+
return fmt.Errorf("QPS limit is set to an invalid value (%f), must be a value in the range [10.0, 1000.0]", qps)
148+
}
149+
*v = LeaderElectionQPSValueWithValidation(qps)
150+
return nil
151+
}
152+
153+
func newLeaderElectionQPSValueWithValidation(defaultVal float64, p *float64) *LeaderElectionQPSValueWithValidation {
154+
*p = defaultVal
155+
return (*LeaderElectionQPSValueWithValidation)(p)
156+
}
157+
158+
type LeaderElectionBurstValueWithValidation int
159+
160+
func (v *LeaderElectionBurstValueWithValidation) String() string {
161+
return fmt.Sprintf("%d", *v)
162+
}
163+
164+
func (v *LeaderElectionBurstValueWithValidation) Set(s string) error {
165+
// Some validation is also performed on the controller manager side and the client-go side. Just
166+
// to be on the safer side we also impose some limits here.
167+
burst, err := strconv.Atoi(s)
168+
if err != nil {
169+
return fmt.Errorf("failed to parse int value: %w", err)
170+
}
171+
172+
if burst < 10 || burst > 2000 {
173+
return fmt.Errorf("burst limit is set to an invalid value (%d), must be a value in the range [10, 2000]", burst)
174+
}
175+
*v = LeaderElectionBurstValueWithValidation(burst)
176+
return nil
177+
}
178+
179+
func newLeaderElectionBurstValueWithValidation(defaultVal int, p *int) *LeaderElectionBurstValueWithValidation {
180+
*p = defaultVal
181+
return (*LeaderElectionBurstValueWithValidation)(p)
98182
}

‎cmd/hubagent/options/options_test.go‎

Lines changed: 72 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -42,11 +42,13 @@ func TestLeaderElectionOpts(t *testing.T) {
4242
flagSetName: "allDefault",
4343
args: []string{},
4444
wantLeaderElectionOpts: LeaderElectionOptions{
45-
LeaderElect: false,
46-
LeaseDuration: metav1.Duration{Duration: 15 * time.Second},
47-
RenewDeadline: metav1.Duration{Duration: 10 * time.Second},
48-
RetryPeriod: metav1.Duration{Duration: 2 * time.Second},
49-
ResourceNamespace: utils.FleetSystemNamespace,
45+
LeaderElect: false,
46+
LeaseDuration: metav1.Duration{Duration: 60 * time.Second},
47+
RenewDeadline: metav1.Duration{Duration: 45 * time.Second},
48+
RetryPeriod: metav1.Duration{Duration: 5 * time.Second},
49+
ResourceNamespace: utils.FleetSystemNamespace,
50+
LeaderElectionQPS: 250.0,
51+
LeaderElectionBurst: 1000,
5052
},
5153
},
5254
{
@@ -58,15 +60,75 @@ func TestLeaderElectionOpts(t *testing.T) {
5860
"--leader-renew-deadline=20s",
5961
"--leader-retry-period=5s",
6062
"--leader-election-namespace=test-namespace",
63+
"--leader-election-qps=500",
64+
"--leader-election-burst=1500",
6165
},
6266
wantLeaderElectionOpts: LeaderElectionOptions{
63-
LeaderElect: true,
64-
LeaseDuration: metav1.Duration{Duration: 30 * time.Second},
65-
RenewDeadline: metav1.Duration{Duration: 20 * time.Second},
66-
RetryPeriod: metav1.Duration{Duration: 5 * time.Second},
67-
ResourceNamespace: "test-namespace",
67+
LeaderElect: true,
68+
LeaseDuration: metav1.Duration{Duration: 30 * time.Second},
69+
RenewDeadline: metav1.Duration{Duration: 20 * time.Second},
70+
RetryPeriod: metav1.Duration{Duration: 5 * time.Second},
71+
ResourceNamespace: "test-namespace",
72+
LeaderElectionQPS: 500.0,
73+
LeaderElectionBurst: 1500,
6874
},
6975
},
76+
{
77+
name: "negative leader election QPS value",
78+
flagSetName: "qpsNegative",
79+
args: []string{"--leader-election-qps=-5"},
80+
wantLeaderElectionOpts: LeaderElectionOptions{
81+
LeaderElect: false,
82+
LeaseDuration: metav1.Duration{Duration: 60 * time.Second},
83+
RenewDeadline: metav1.Duration{Duration: 45 * time.Second},
84+
RetryPeriod: metav1.Duration{Duration: 5 * time.Second},
85+
ResourceNamespace: utils.FleetSystemNamespace,
86+
LeaderElectionQPS: -1,
87+
LeaderElectionBurst: 1000,
88+
},
89+
},
90+
{
91+
name: "leader election QPS parse error",
92+
flagSetName: "qpsParseError",
93+
args: []string{"--leader-election-qps=abc"},
94+
wantErred: true,
95+
wantErrMsgSubStr: "failed to parse float64 value",
96+
},
97+
{
98+
name: "leader election QPS out of range (too small)",
99+
flagSetName: "qpsOutOfRangeTooSmall",
100+
args: []string{"--leader-election-qps=9.9"},
101+
wantErred: true,
102+
wantErrMsgSubStr: "QPS limit is set to an invalid value",
103+
},
104+
{
105+
name: "leader election QPS out of range (too large)",
106+
flagSetName: "qpsOutOfRangeTooLarge",
107+
args: []string{"--leader-election-qps=1000.1"},
108+
wantErred: true,
109+
wantErrMsgSubStr: "QPS limit is set to an invalid value",
110+
},
111+
{
112+
name: "leader election burst parse error",
113+
flagSetName: "burstParseError",
114+
args: []string{"--leader-election-burst=abc"},
115+
wantErred: true,
116+
wantErrMsgSubStr: "failed to parse int value",
117+
},
118+
{
119+
name: "leader election burst out of range (too small)",
120+
flagSetName: "burstOutOfRangeTooSmall",
121+
args: []string{"--leader-election-burst=9"},
122+
wantErred: true,
123+
wantErrMsgSubStr: "burst limit is set to an invalid value",
124+
},
125+
{
126+
name: "leader election burst out of range (too large)",
127+
flagSetName: "burstOutOfRangeTooLarge",
128+
args: []string{"--leader-election-burst=2001"},
129+
wantErred: true,
130+
wantErrMsgSubStr: "burst limit is set to an invalid value",
131+
},
70132
}
71133

72134
for _, tc := range testCases {

‎cmd/hubagent/options/validation.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,11 @@ func (o *Options) Validate() field.ErrorList {
3333
errs = append(errs, field.Invalid(newPath.Child("HubBurst"), o.CtrlMgrOpts.HubBurst, "The burst limit for client-side throttling must be greater than or equal to its QPS limit"))
3434
}
3535

36+
// Cross-field validation for leader election options.
37+
if float64(o.LeaderElectionOpts.LeaderElectionBurst) < float64(o.LeaderElectionOpts.LeaderElectionQPS) {
38+
errs = append(errs, field.Invalid(newPath.Child("LeaderElectionBurst"), o.LeaderElectionOpts.LeaderElectionBurst, "The burst limit for client-side throttling of leader election related operations must be greater than or equal to its QPS limit"))
39+
}
40+
3641
// Cross-field validation for webhook options.
3742

3843
// Note: this validation logic is a bit weird in the sense that the system accepts

‎cmd/hubagent/options/validation_test.go‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,10 @@ func newTestOptions(modifyOptions ModifyOptions) Options {
4040
HubQPS: 250,
4141
HubBurst: 1000,
4242
},
43+
LeaderElectionOpts: LeaderElectionOptions{
44+
LeaderElectionQPS: 250.0,
45+
LeaderElectionBurst: 1000,
46+
},
4347
WebhookOpts: WebhookOptions{
4448
ClientConnectionType: "url",
4549
ServiceName: testWebhookServiceName,
@@ -78,6 +82,13 @@ func TestValidateControllerManagerConfiguration(t *testing.T) {
7882
}),
7983
want: field.ErrorList{field.Invalid(newPath.Child("HubBurst"), 50, "The burst limit for client-side throttling must be greater than or equal to its QPS limit")},
8084
},
85+
"invalid: leader election burst value is less than its QPS value": {
86+
opt: newTestOptions(func(option *Options) {
87+
option.LeaderElectionOpts.LeaderElectionQPS = 100
88+
option.LeaderElectionOpts.LeaderElectionBurst = 50
89+
}),
90+
want: field.ErrorList{field.Invalid(newPath.Child("LeaderElectionBurst"), 50, "The burst limit for client-side throttling of leader election related operations must be greater than or equal to its QPS limit")},
91+
},
8192
"WebhookServiceName is empty": {
8293
opt: newTestOptions(func(option *Options) {
8394
option.WebhookOpts.EnableWebhooks = true

0 commit comments

Comments
 (0)