Mirrors/flux2
1
0
Fork 0
mirror of synced 2026-02-06 19:05:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add the kube client qps and burst to the global args

Browse Source 
Signed-off-by: Stefan Prodan <stefan.prodan@gmail.com>
This commit is contained in:
Stefan Prodan
2022-03-24 20:02:17 +02:00
committed by Hidde Beydals
parent 150d9d7ae6
commit 0d8194c800
46 changed files with 114 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
internal/utils/apply.go
View File

@@ -25,19 +25,21 @@ import (
"path/filepath"
"time"
"github.com/fluxcd/pkg/ssa"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/cli-runtime/pkg/genericclioptions"
"sigs.k8s.io/cli-utils/pkg/kstatus/polling"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/kustomize/api/konfig"
runclient "github.com/fluxcd/pkg/runtime/client"
"github.com/fluxcd/pkg/ssa"
"github.com/fluxcd/flux2/pkg/manifestgen/kustomization"
)
// Apply is the equivalent of 'kubectl apply --server-side -f'.
// If the given manifest is a kustomization.yaml, then apply performs the equivalent of 'kubectl apply --server-side -k'.
func Apply(ctx context.Context, rcg genericclioptions.RESTClientGetter, manifestPath string) (string, error) {
func Apply(ctx context.Context, rcg genericclioptions.RESTClientGetter, opts *runclient.Options, manifestPath string) (string, error) {
objs, err := readObjects(manifestPath)
if err != nil {
return "", err
@@ -68,19 +70,19 @@ func Apply(ctx context.Context, rcg genericclioptions.RESTClientGetter, manifest
}
if len(stageOne) > 0 {
cs, err := applySet(ctx, rcg, stageOne)
cs, err := applySet(ctx, rcg, opts, stageOne)
if err != nil {
return "", err
}
changeSet.Append(cs.Entries)
}
if err := waitForSet(rcg, changeSet); err != nil {
if err := waitForSet(rcg, opts, changeSet); err != nil {
return "", err
}
if len(stageTwo) > 0 {
cs, err := applySet(ctx, rcg, stageTwo)
cs, err := applySet(ctx, rcg, opts, stageTwo)
if err != nil {
return "", err
}
@@ -112,8 +114,8 @@ func readObjects(manifestPath string) ([]*unstructured.Unstructured, error) {
return ssa.ReadObjects(bufio.NewReader(ms))
}
func newManager(rcg genericclioptions.RESTClientGetter) (*ssa.ResourceManager, error) {
cfg, err := KubeConfig(rcg)
func newManager(rcg genericclioptions.RESTClientGetter, opts *runclient.Options) (*ssa.ResourceManager, error) {
cfg, err := KubeConfig(rcg, opts)
if err != nil {
return nil, err
}
@@ -134,8 +136,8 @@ func newManager(rcg genericclioptions.RESTClientGetter) (*ssa.ResourceManager, e
}
func applySet(ctx context.Context, rcg genericclioptions.RESTClientGetter, objects []*unstructured.Unstructured) (*ssa.ChangeSet, error) {
man, err := newManager(rcg)
func applySet(ctx context.Context, rcg genericclioptions.RESTClientGetter, opts *runclient.Options, objects []*unstructured.Unstructured) (*ssa.ChangeSet, error) {
man, err := newManager(rcg, opts)
if err != nil {
return nil, err
}
@@ -143,8 +145,8 @@ func applySet(ctx context.Context, rcg genericclioptions.RESTClientGetter, objec
return man.ApplyAll(ctx, objects, ssa.DefaultApplyOptions())
}
func waitForSet(rcg genericclioptions.RESTClientGetter, changeSet *ssa.ChangeSet) error {
man, err := newManager(rcg)
func waitForSet(rcg genericclioptions.RESTClientGetter, opts *runclient.Options, changeSet *ssa.ChangeSet) error {
man, err := newManager(rcg, opts)
if err != nil {
return err
}

12
internal/utils/utils.go
View File

@@ -47,6 +47,7 @@ import (
kustomizev1 "github.com/fluxcd/kustomize-controller/api/v1beta2"
notificationv1 "github.com/fluxcd/notification-controller/api/v1beta1"
"github.com/fluxcd/pkg/apis/meta"
runclient "github.com/fluxcd/pkg/runtime/client"
"github.com/fluxcd/pkg/version"
sourcev1 "github.com/fluxcd/source-controller/api/v1beta2"
@@ -106,15 +107,15 @@ func ExecKubectlCommand(ctx context.Context, mode ExecMode, kubeConfigPath strin
return "", nil
}
func KubeConfig(rcg genericclioptions.RESTClientGetter) (*rest.Config, error) {
func KubeConfig(rcg genericclioptions.RESTClientGetter, opts *runclient.Options) (*rest.Config, error) {
cfg, err := rcg.ToRESTConfig()
if err != nil {
return nil, fmt.Errorf("kubernetes configuration load failed: %w", err)
}
// avoid throttling request when some Flux CRDs are not registered
cfg.QPS = 50
cfg.Burst = 100
cfg.QPS = opts.QPS
cfg.Burst = opts.Burst
return cfg, nil
}
@@ -137,12 +138,15 @@ func NewScheme() *apiruntime.Scheme {
return scheme
}
func KubeClient(rcg genericclioptions.RESTClientGetter) (client.WithWatch, error) {
func KubeClient(rcg genericclioptions.RESTClientGetter, opts *runclient.Options) (client.WithWatch, error) {
cfg, err := rcg.ToRESTConfig()
if err != nil {
return nil, err
}
cfg.QPS = opts.QPS
cfg.Burst = opts.Burst
scheme := NewScheme()
kubeClient, err := client.NewWithWatch(cfg, client.Options{
Scheme: scheme,