From 5c98b6af69486d1aa2fe37f28b010a403103cf04 Mon Sep 17 00:00:00 2001
From: David Vallee Delisle <me@dvd.dev>
Date: Mon, 11 Aug 2025 16:21:56 -0400
Subject: [PATCH] k8s_locker: add configurable QPS and Burst for the Kubernetes
 API client

The k8s_locker uses the default Kubernetes client QPS (5) and Burst (10)
inherited from rest.InClusterConfig(). Under high-churn conditions or
large clusters, these limits cause the locker to back-pressure against
the API server, which can manifest as intermittent watch errors when the
lease-renewal and peer-discovery goroutines compete for the same rate
bucket.

Add two optional config fields:

  locker:
    type: k8s
    qps: 100    # max sustained requests/s to the k8s API (default: unchanged)
    burst: 100  # max burst above QPS (default: unchanged)

When either field is zero or absent, the in-cluster default is kept, so
the change is fully backward-compatible.

Resolves: intermittent 'k8s_locker watch ended with error' log spam in
multi-replica deployments where lease + EndpointSlice watches share a
single rate-limited client.
---
 pkg/lockers/k8s_locker/k8s_locker.go | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/pkg/lockers/k8s_locker/k8s_locker.go b/pkg/lockers/k8s_locker/k8s_locker.go
index 073fedc32..6e7bb4893 100644
--- a/pkg/lockers/k8s_locker/k8s_locker.go
+++ b/pkg/lockers/k8s_locker/k8s_locker.go
@@ -67,6 +67,8 @@ type config struct {
 	RenewPeriod   time.Duration `mapstructure:"renew-period,omitempty" json:"renew-period,omitempty"`
 	RetryTimer    time.Duration `mapstructure:"retry-timer,omitempty" json:"retry-timer,omitempty"`
 	Debug         bool          `mapstructure:"debug,omitempty" json:"debug,omitempty"`
+	QPS           float32       `mapstructure:"qps,omitempty" json:"qps,omitempty"`
+	Burst         int           `mapstructure:"burst,omitempty" json:"burst,omitempty"`
 }
 
 type lock struct {
@@ -91,6 +93,13 @@ func (k *k8sLocker) Init(ctx context.Context, cfg map[string]interface{}, opts .
 		return err
 	}
 
+	if k.Cfg.QPS > 0 {
+		inClusterConfig.QPS = k.Cfg.QPS
+	}
+	if k.Cfg.Burst > 0 {
+		inClusterConfig.Burst = k.Cfg.Burst
+	}
+
 	k.clientset, err = kubernetes.NewForConfig(inClusterConfig)
 	if err != nil {
 		return err