Merge pull request #375 from werf/feat/optimize-kube-api-load

feat: greatly decrease Kubernetes apiserver load

Author: ilya-lesikov

go.mod

@@ -1,8 +1,6 @@
 module github.com/werf/kubedog
 
-go 1.21
-
-toolchain go1.21.6
+go 1.23.0
 
 require (
 	github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d

pkg/informer/factory.go

@@ -0,0 +1,65 @@
+package informer
+
+import (
+	"fmt"
+	"sync"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/dynamic/dynamicinformer"
+
+	"github.com/werf/kubedog/pkg/trackers/dyntracker/util"
+)
+
+type ConcurrentInformerFactoryOptions struct{}
+
+func NewConcurrentInformerFactory(stopCh <-chan struct{}, watchErrCh chan<- error, dynamicClient dynamic.Interface, opts ConcurrentInformerFactoryOptions) *util.Concurrent[*InformerFactory] {
+	lock := &sync.RWMutex{}
+	return util.NewConcurrentWithLock(&InformerFactory{
+		dynamicClient:       dynamicClient,
+		namespacedFactories: make(map[string]dynamicinformer.DynamicSharedInformerFactory),
+		informersLock:       lock,
+		stopCh:              stopCh,
+		watchErrCh:          watchErrCh,
+	}, lock)
+}
+
+type InformerFactory struct {
+	clusteredFactory    dynamicinformer.DynamicSharedInformerFactory
+	dynamicClient       dynamic.Interface
+	informersLock       *sync.RWMutex
+	namespacedFactories map[string]dynamicinformer.DynamicSharedInformerFactory
+	stopCh              <-chan struct{}
+	watchErrCh          chan<- error
+}
+
+func (f *InformerFactory) ForNamespace(gvr schema.GroupVersionResource, namespace string) (*util.Concurrent[*Informer], error) {
+	factory, found := f.namespacedFactories[namespace]
+	if !found {
+		factory = dynamicinformer.NewFilteredDynamicSharedInformerFactory(f.dynamicClient, 0, namespace, nil)
+		f.namespacedFactories[namespace] = factory
+	}
+
+	informer, err := newInformerFromFactory(gvr, factory, f.stopCh, f.watchErrCh, informerFromFactoryOptions{
+		Namespace: namespace,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("construct informer: %w", err)
+	}
+
+	return util.NewConcurrentWithLock(informer, f.informersLock), nil
+}
+
+func (f *InformerFactory) Clustered(gvr schema.GroupVersionResource) (*util.Concurrent[*Informer], error) {
+	if f.clusteredFactory == nil {
+		f.clusteredFactory = dynamicinformer.NewFilteredDynamicSharedInformerFactory(f.dynamicClient, 0, metav1.NamespaceAll, nil)
+	}
+
+	informer, err := newInformerFromFactory(gvr, f.clusteredFactory, f.stopCh, f.watchErrCh, informerFromFactoryOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("construct informer: %w", err)
+	}
+
+	return util.NewConcurrentWithLock(informer, f.informersLock), nil
+}

pkg/informer/informer.go

@@ -0,0 +1,109 @@
+package informer
+
+import (
+	"fmt"
+	"io"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/dynamic/dynamicinformer"
+	"k8s.io/client-go/tools/cache"
+
+	"github.com/werf/kubedog/pkg/tracker/debug"
+)
+
+type informerFromFactoryOptions struct {
+	Namespace string
+}
+
+func newInformerFromFactory(gvr schema.GroupVersionResource, factory dynamicinformer.DynamicSharedInformerFactory, stopCh <-chan struct{}, watchErrCh chan<- error, opts informerFromFactoryOptions) (*Informer, error) {
+	informer := factory.ForResource(gvr)
+
+	if err := setWatchErrorHandler(informer.Informer().SetWatchErrorHandler, watchErrCh, gvr); err != nil {
+		return nil, fmt.Errorf("set watch error handler for resource %s: %w", gvr.String(), err)
+	}
+
+	return &Informer{
+		factory:   factory,
+		informer:  informer.Informer(),
+		lister:    informer.Lister(),
+		namespace: opts.Namespace,
+		stopCh:    stopCh,
+	}, nil
+}
+
+type Informer struct {
+	factory   dynamicinformer.DynamicSharedInformerFactory
+	informer  cache.SharedIndexInformer
+	lister    cache.GenericLister
+	namespace string
+	stopCh    <-chan struct{}
+}
+
+func (i *Informer) AddEventHandler(handler cache.ResourceEventHandler) (cache.ResourceEventHandlerRegistration, error) {
+	return i.informer.AddEventHandler(handler)
+}
+
+func (i *Informer) RemoveEventHandler(handle cache.ResourceEventHandlerRegistration) error {
+	return i.informer.RemoveEventHandler(handle)
+}
+
+func (i *Informer) Run() {
+	i.factory.Start(i.stopCh)
+}
+
+func (i *Informer) List(selector labels.Selector) (ret []runtime.Object, err error) {
+	if i.namespace != "" {
+		return i.lister.ByNamespace(i.namespace).List(selector)
+	}
+
+	return i.lister.List(selector)
+}
+
+func (i *Informer) Get(name string) (runtime.Object, error) {
+	if i.namespace != "" {
+		return i.lister.ByNamespace(i.namespace).Get(name)
+	}
+
+	return i.lister.Get(name)
+}
+
+func setWatchErrorHandler(setWatchErrorHandler func(handler cache.WatchErrorHandler) error, watchErrCh chan<- error, gvr schema.GroupVersionResource) error {
+	if err := setWatchErrorHandler(
+		func(r *cache.Reflector, err error) {
+			isExpiredError := func(err error) bool {
+				return apierrors.IsResourceExpired(err) || apierrors.IsGone(err)
+			}
+
+			// Based on: k8s.io/client-go@v0.30.11/tools/cache/reflector.go
+			switch {
+			case isExpiredError(err):
+				if debug.Debug() {
+					fmt.Printf("[SetWatchErrorHandler] %s watch closed with expired error: %s\n", gvr.String(), err)
+				}
+			case err == io.EOF:
+				// watch closed normally
+			case err == io.ErrUnexpectedEOF:
+				if debug.Debug() {
+					fmt.Printf("[SetWatchErrorHandler] %s watch closed with unexpected EOF error: %s\n", gvr.String(), err)
+				}
+			default:
+				if debug.Debug() {
+					fmt.Printf("[SetWatchErrorHandler] %s watch closed with an error: %s\n", gvr.String(), err)
+				}
+
+				watchErrCh <- fmt.Errorf("unrecoverable watch error for %s: %w", gvr.String(), err)
+			}
+		},
+	); err != nil {
+		if err.Error() == "informer has already started" {
+			return nil
+		}
+
+		return err
+	}
+
+	return nil
+}

pkg/tracker/canary/feed.go

@@ -67,7 +67,7 @@ func (f *feed) Track(name, namespace string, kube kubernetes.Interface, opts tra
 	ctx, cancel := watchtools.ContextWithOptionalTimeout(parentContext, opts.Timeout)
 	defer cancel()
 
-	canary := NewTracker(name, namespace, kube, nil, opts)
+	canary := NewTracker(name, namespace, kube, nil, nil, opts)
 
 	go func() {
 		err := canary.Track(ctx)

pkg/tracker/canary/tracker.go

@@ -6,22 +6,26 @@ import (
 	"time"
 
 	"github.com/fluxcd/flagger/pkg/apis/flagger/v1beta1"
-	flaggerv1beta1 "github.com/fluxcd/flagger/pkg/client/clientset/versioned/typed/flagger/v1beta1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/fields"
+	flaggerscheme "github.com/fluxcd/flagger/pkg/client/clientset/versioned/scheme"
+	"github.com/samber/lo"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/tools/cache"
-	watchtools "k8s.io/client-go/tools/watch"
 
-	"github.com/werf/kubedog/pkg/kube"
+	"github.com/werf/kubedog/pkg/informer"
 	"github.com/werf/kubedog/pkg/tracker"
 	"github.com/werf/kubedog/pkg/tracker/debug"
+	"github.com/werf/kubedog/pkg/trackers/dyntracker/util"
 )
 
+func init() {
+	flaggerscheme.AddToScheme(scheme.Scheme)
+}
+
 type FailedReport struct {
 	FailedReason string
 	CanaryStatus CanaryStatus
@@ -54,14 +58,15 @@ type Tracker struct {
 	dynamicClient dynamic.Interface
 }
 
-func NewTracker(name, namespace string, kube kubernetes.Interface, dynamicClient dynamic.Interface, opts tracker.Options) *Tracker {
+func NewTracker(name, namespace string, kube kubernetes.Interface, dynamicClient dynamic.Interface, informerFactory *util.Concurrent[*informer.InformerFactory], opts tracker.Options) *Tracker {
 	return &Tracker{
 		Tracker: tracker.Tracker{
 			Kube:             kube,
 			Namespace:        namespace,
 			FullResourceName: fmt.Sprintf("canary/%s", name),
 			ResourceName:     name,
 			LogsFromTime:     opts.LogsFromTime,
+			InformerFactory:  informerFactory,
 		},
 
 		Added:     make(chan CanaryStatus, 1),
@@ -84,10 +89,12 @@ func NewTracker(name, namespace string, kube kubernetes.Interface, dynamicClient
 }
 
 func (canary *Tracker) Track(ctx context.Context) error {
-	err := canary.runInformer(ctx)
+	canaryInformerCleanupFn, err := canary.runInformer(ctx)
 	if err != nil {
 		return err
 	}
+	defer canaryInformerCleanupFn()
+
 	for {
 		select {
 		case object := <-canary.objectAdded:
@@ -123,85 +130,67 @@ func (canary *Tracker) Track(ctx context.Context) error {
 	}
 }
 
-func (canary *Tracker) runInformer(ctx context.Context) error {
-	tweakListOptions := func(options metav1.ListOptions) metav1.ListOptions {
-		options.FieldSelector = fields.OneTermEqualSelector("metadata.name", canary.ResourceName).String()
-		return options
-	}
-
-	var lw *cache.ListWatch
-	if canary.dynamicClient != nil {
-		lw = &cache.ListWatch{
-			ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
-				return canary.dynamicClient.Resource(schema.GroupVersionResource{
-					Group:    "flagger.app",
-					Version:  "v1beta1",
-					Resource: "canaries",
-				}).Namespace(canary.Namespace).List(ctx, tweakListOptions(options))
-			},
-			WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
-				return canary.dynamicClient.Resource(schema.GroupVersionResource{
-					Group:    "flagger.app",
-					Version:  "v1beta1",
-					Resource: "canaries",
-				}).Namespace(canary.Namespace).Watch(ctx, tweakListOptions(options))
-			},
-		}
-	} else {
-		config, err := kube.GetKubeConfig(kube.KubeConfigOptions{})
+func (canary *Tracker) runInformer(ctx context.Context) (cleanupFn func(), err error) {
+	var inform *util.Concurrent[*informer.Informer]
+	if err := canary.InformerFactory.RWTransactionErr(func(factory *informer.InformerFactory) error {
+		inform, err = factory.ForNamespace(schema.GroupVersionResource{
+			Group:    "flagger.app",
+			Version:  "v1beta1",
+			Resource: "canaries",
+		}, canary.Namespace)
 		if err != nil {
-			fmt.Print(err)
+			return fmt.Errorf("get informer from factory: %w", err)
 		}
 
-		flagger, err := flaggerv1beta1.NewForConfig(config.Config)
-		if err != nil {
-			fmt.Print(err)
-		}
+		return nil
+	}); err != nil {
+		return nil, err
+	}
 
-		lw = &cache.ListWatch{
-			ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
-				return flagger.Canaries(canary.Namespace).List(ctx, tweakListOptions(options))
-			},
-			WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
-				return flagger.Canaries(canary.Namespace).Watch(ctx, tweakListOptions(options))
+	if err := inform.RWTransactionErr(func(inf *informer.Informer) error {
+		handler, err := inf.AddEventHandler(
+			cache.FilteringResourceEventHandler{
+				FilterFunc: func(obj interface{}) bool {
+					canaryObj := &v1beta1.Canary{}
+					lo.Must0(runtime.DefaultUnstructuredConverter.FromUnstructured(obj.(*unstructured.Unstructured).Object, canaryObj))
+					return canaryObj.Name == canary.ResourceName &&
+						canaryObj.Namespace == canary.Namespace
+				},
+				Handler: cache.ResourceEventHandlerFuncs{
+					AddFunc: func(obj interface{}) {
+						canaryObj := &v1beta1.Canary{}
+						lo.Must0(runtime.DefaultUnstructuredConverter.FromUnstructured(obj.(*unstructured.Unstructured).Object, canaryObj))
+						canary.objectAdded <- canaryObj
+					},
+					UpdateFunc: func(oldObj, newObj interface{}) {
+						canaryObj := &v1beta1.Canary{}
+						lo.Must0(runtime.DefaultUnstructuredConverter.FromUnstructured(newObj.(*unstructured.Unstructured).Object, canaryObj))
+						canary.objectModified <- canaryObj
+					},
+					DeleteFunc: func(obj interface{}) {
+						canaryObj := &v1beta1.Canary{}
+						lo.Must0(runtime.DefaultUnstructuredConverter.FromUnstructured(obj.(*unstructured.Unstructured).Object, canaryObj))
+						canary.objectDeleted <- canaryObj
+					},
+				},
 			},
+		)
+		if err != nil {
+			return fmt.Errorf("add event handler: %w", err)
 		}
-	}
-
-	go func() {
-		_, err := watchtools.UntilWithSync(ctx, lw, &v1beta1.Canary{}, nil, func(e watch.Event) (bool, error) {
-			if debug.Debug() {
-				fmt.Printf("Canary `%s` informer event: %#v\n", canary.ResourceName, e.Type)
-			}
-
-			var object *v1beta1.Canary
-
-			if e.Type != watch.Error {
-				var ok bool
-				object, ok = e.Object.(*v1beta1.Canary)
-				if !ok {
-					return true, fmt.Errorf("expected %s to be a *v1beta1.Canary, got %T", canary.ResourceName, e.Object)
-				}
-			}
 
-			switch e.Type {
-			case watch.Added:
-				canary.objectAdded <- object
-			case watch.Modified:
-				canary.objectModified <- object
-			case watch.Deleted:
-				canary.objectDeleted <- object
-			}
+		cleanupFn = func() {
+			inf.RemoveEventHandler(handler)
+		}
 
-			return false, nil
-		})
+		inf.Run()
 
-		if err != tracker.AdaptInformerError(err) {
-			canary.errors <- fmt.Errorf("canary informer error: %w", err)
-		}
-	}()
+		return nil
+	}); err != nil {
+		return nil, err
+	}
 
-	return nil
+	return cleanupFn, nil
 }
 
 func (canary *Tracker) handleCanaryState(ctx context.Context, object *v1beta1.Canary) error {