Skip to content

Commit e002b89

Browse files
plaffittnpdgm
plaffitt
authored and
npdgm
committed
feat: configurable burst and QPS in k8s client
1 parent cc3b0c0 commit e002b89

File tree

5 files changed

+34
-9
lines changed

5 files changed

+34
-9
lines changed

‎cmd/x509-certificate-exporter/main.go

+12-1
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ import (
1212

1313
_ "github.com/KimMachineGun/automemlimit"
1414
_ "go.uber.org/automaxprocs"
15+
"k8s.io/client-go/util/flowcontrol"
1516

1617
"github.com/enix/x509-certificate-exporter/v3/internal"
1718
getopt "github.com/pborman/getopt/v2"
@@ -39,6 +40,9 @@ func main() {
3940
maxCacheDuration := durationFlag(0)
4041
getopt.FlagLong(&maxCacheDuration, "max-cache-duration", 0, "maximum cache duration for kube secrets. cache is per namespace and randomized to avoid massive requests.")
4142

43+
rateLimitQPS := getopt.IntLong("kube-api-rate-limit-qps", 0, 0, "Kubernetes API request rate limit")
44+
rateLimitBurst := getopt.IntLong("kube-api-rate-limit-burst", 0, 0, "Kubernetes API request burst")
45+
4246
files := stringArrayFlag{}
4347
getopt.FlagLong(&files, "watch-file", 'f', "watch one or more x509 certificate file")
4448

@@ -131,7 +135,14 @@ func main() {
131135
configpath = defaultKubeConfig
132136
}
133137

134-
err := exporter.ConnectToKubernetesCluster(configpath)
138+
// Set rate limiter only if both QPS and burst are set
139+
var rateLimiter flowcontrol.RateLimiter
140+
if *rateLimitQPS > 0 && *rateLimitBurst > 0 {
141+
log.Infof("setting Kubernetes API rate limiter to %d QPS and %d burst", *rateLimitQPS, *rateLimitBurst)
142+
rateLimiter = flowcontrol.NewTokenBucketRateLimiter(float32(*rateLimitQPS), *rateLimitBurst)
143+
}
144+
145+
err := exporter.ConnectToKubernetesCluster(configpath, rateLimiter)
135146
if err != nil {
136147
log.Fatal(err)
137148
}

‎deploy/charts/x509-certificate-exporter/templates/deployment.yaml

+4
Original file line numberDiff line numberDiff line change
@@ -125,6 +125,10 @@ spec:
125125
{{- else }}
126126
- --max-cache-duration=0
127127
{{- end }}
128+
{{- with .Values.secretsExporter.kubeApiRateLimits }}
129+
- --kube-api-rate-limit-qps={{ .qps }}
130+
- --kube-api-rate-limit-burst={{ .burst }}
131+
{{- end }}
128132
{{- if .Values.exposePerCertificateErrorMetrics }}
129133
- --expose-per-cert-error-metrics
130134
{{- end }}

‎deploy/charts/x509-certificate-exporter/values.yaml

+5
Original file line numberDiff line numberDiff line change
@@ -128,6 +128,11 @@ secretsExporter:
128128
# -- Maximum time an object can stay in cache unrefreshed (seconds) - it will be at least half of that
129129
maxDuration: 300
130130

131+
kubeApiRateLimits: {}
132+
# -- Try higher values if querying secrets takes a long time because of throttling
133+
# qps: 5
134+
# burst: 10
135+
131136
# -- Additional environment variables for container
132137
env: []
133138
# - name: GOMAXPROCS

‎internal/kubernetes.go

+8-3
Original file line numberDiff line numberDiff line change
@@ -15,13 +15,14 @@ import (
1515
"k8s.io/client-go/kubernetes"
1616
"k8s.io/client-go/rest"
1717
"k8s.io/client-go/tools/clientcmd"
18+
"k8s.io/client-go/util/flowcontrol"
1819
)
1920

2021
// ConnectToKubernetesCluster : Try connect to a cluster from inside if path is empty,
2122
// otherwise try loading the kubeconfig at path "path"
22-
func (exporter *Exporter) ConnectToKubernetesCluster(path string) error {
23+
func (exporter *Exporter) ConnectToKubernetesCluster(path string, rateLimiter flowcontrol.RateLimiter) error {
2324
var err error
24-
exporter.kubeClient, err = connectToKubernetesCluster(path, false)
25+
exporter.kubeClient, err = connectToKubernetesCluster(path, false, rateLimiter)
2526
return err
2627
}
2728

@@ -232,7 +233,7 @@ func (exporter *Exporter) shrinkSecret(secret v1.Secret) v1.Secret {
232233
return secret
233234
}
234235

235-
func connectToKubernetesCluster(kubeconfigPath string, insecure bool) (*kubernetes.Clientset, error) {
236+
func connectToKubernetesCluster(kubeconfigPath string, insecure bool, rateLimiter flowcontrol.RateLimiter) (*kubernetes.Clientset, error) {
236237
config, err := parseKubeConfig(kubeconfigPath)
237238
if err != nil {
238239
return nil, err
@@ -243,6 +244,10 @@ func connectToKubernetesCluster(kubeconfigPath string, insecure bool) (*kubernet
243244
config.TLSClientConfig.CAData = nil
244245
}
245246

247+
if rateLimiter != nil {
248+
config.RateLimiter = rateLimiter
249+
}
250+
246251
return getKubeClient(config)
247252
}
248253

‎internal/kubernetes_test.go

+5-5
Original file line numberDiff line numberDiff line change
@@ -54,7 +54,7 @@ func TestMain(m *testing.M) {
5454
panic(err)
5555
}
5656

57-
sharedKubeClient, err = connectToKubernetesCluster("kubeconfig", true)
57+
sharedKubeClient, err = connectToKubernetesCluster("kubeconfig", true, nil)
5858
if err != nil {
5959
panic(err)
6060
}
@@ -286,7 +286,7 @@ func TestKubeMetricLabels(t *testing.T) {
286286
}
287287

288288
func TestKubeNamespaceListFailure(t *testing.T) {
289-
kubeClient, err := connectToKubernetesCluster("kubeconfig.x509-certificate-exporter", true)
289+
kubeClient, err := connectToKubernetesCluster("kubeconfig.x509-certificate-exporter", true, nil)
290290
if err != nil {
291291
panic(err)
292292
}
@@ -301,7 +301,7 @@ func TestKubeNamespaceListFailure(t *testing.T) {
301301
}
302302

303303
func TestKubeSecretsListFailure(t *testing.T) {
304-
kubeClient, err := connectToKubernetesCluster("kubeconfig.x509-certificate-exporter-list", true)
304+
kubeClient, err := connectToKubernetesCluster("kubeconfig.x509-certificate-exporter-list", true, nil)
305305
if err != nil {
306306
panic(err)
307307
}
@@ -316,7 +316,7 @@ func TestKubeSecretsListFailure(t *testing.T) {
316316
}
317317

318318
func TestKubeInvalidConfig(t *testing.T) {
319-
_, err := connectToKubernetesCluster("../test/kubeconfig-corrupted.yml", true)
319+
_, err := connectToKubernetesCluster("../test/kubeconfig-corrupted.yml", true, nil)
320320
assert.NotNil(t, err)
321321
}
322322

@@ -370,7 +370,7 @@ func TestKubeEmptyStringKey(t *testing.T) {
370370

371371
func TestKubeConnectionFromInsideFailure(t *testing.T) {
372372
e := &Exporter{}
373-
err := e.ConnectToKubernetesCluster("")
373+
err := e.ConnectToKubernetesCluster("", nil)
374374
assert.NotNil(t, err)
375375
}
376376

0 commit comments

Comments
 (0)