contextstore

Author: mjudeikis

pkg/konnector/controllers/cluster/serviceexport/serviceexport_controller.go

@@ -35,6 +35,7 @@ import (
 
 	"github.com/kube-bind/kube-bind/pkg/committer"
 	"github.com/kube-bind/kube-bind/pkg/indexers"
+	"github.com/kube-bind/kube-bind/pkg/konnector/controllers/contextstore"
 	"github.com/kube-bind/kube-bind/pkg/konnector/controllers/dynamic"
 	kubebindv1alpha2 "github.com/kube-bind/kube-bind/sdk/apis/kubebind/v1alpha2"
 	bindclient "github.com/kube-bind/kube-bind/sdk/client/clientset/versioned"
@@ -94,7 +95,7 @@ func NewController(
 			consumerConfig:           consumerConfig,
 			providerConfig:           providerConfig,
 
-			syncContext: map[string]syncContext{},
+			syncStore: contextstore.New(),
 
 			getServiceBinding: func(name string) (*kubebindv1alpha2.APIServiceBinding, error) {
 				return serviceBindingInformer.Lister().Get(name)
@@ -287,27 +288,27 @@ func (c *controller) processNextWorkItem(ctx context.Context) bool {
 }
 
 func (c *controller) process(ctx context.Context, key string) error {
-	ns, name, err := cache.SplitMetaNamespaceKey(key)
+	namespace, name, err := cache.SplitMetaNamespaceKey(key)
 	if err != nil {
 		runtime.HandleError(err)
 		return nil // we cannot do anything
 	}
 
 	logger := klog.FromContext(ctx)
 
-	obj, err := c.serviceExportLister.APIServiceExports(ns).Get(name)
+	obj, err := c.serviceExportLister.APIServiceExports(namespace).Get(name)
 	if err != nil && !errors.IsNotFound(err) {
 		return err
 	} else if errors.IsNotFound(err) {
 		logger.Error(err, "APIServiceExport disappeared")
-		return c.reconcile(ctx, name, nil)
+		return c.reconcile(ctx, namespace, name, nil)
 	}
 
 	old := obj
 	obj = obj.DeepCopy()
 
 	var errs []error
-	if err := c.reconcile(ctx, name, obj); err != nil {
+	if err := c.reconcile(ctx, namespace, name, obj); err != nil {
 		errs = append(errs, err)
 	}
 

pkg/konnector/controllers/cluster/serviceexport/serviceexport_reconcile.go

@@ -22,7 +22,6 @@ import (
 	"maps"
 	"slices"
 	"strings"
-	"sync"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
@@ -41,6 +40,7 @@ import (
 	"github.com/kube-bind/kube-bind/pkg/konnector/controllers/cluster/serviceexport/multinsinformer"
 	"github.com/kube-bind/kube-bind/pkg/konnector/controllers/cluster/serviceexport/spec"
 	"github.com/kube-bind/kube-bind/pkg/konnector/controllers/cluster/serviceexport/status"
+	"github.com/kube-bind/kube-bind/pkg/konnector/controllers/contextstore"
 	"github.com/kube-bind/kube-bind/pkg/konnector/controllers/dynamic"
 	kubebindv1alpha2 "github.com/kube-bind/kube-bind/sdk/apis/kubebind/v1alpha2"
 	conditionsapi "github.com/kube-bind/kube-bind/sdk/apis/third_party/conditions/apis/conditions/v1alpha1"
@@ -56,8 +56,7 @@ type reconciler struct {
 
 	consumerConfig, providerConfig *rest.Config
 
-	lock        sync.Mutex
-	syncContext map[string]syncContext // by CRD name
+	syncStore contextstore.Store // by APIServiceExport name. This includes same ctx for resourc claims and crds.
 
 	getCRD               func(name string) (*apiextensionsv1.CustomResourceDefinition, error)
 	getServiceBinding    func(name string) (*kubebindv1alpha2.APIServiceBinding, error)
@@ -66,15 +65,10 @@ type reconciler struct {
 	updateRemoteBoundSchema func(ctx context.Context, boundSchema *kubebindv1alpha2.BoundSchema) error
 }
 
-type syncContext struct {
-	generation int64
-	cancel     func()
-}
-
-func (r *reconciler) reconcile(ctx context.Context, name string, export *kubebindv1alpha2.APIServiceExport) error {
+func (r *reconciler) reconcile(ctx context.Context, namespace, name string, export *kubebindv1alpha2.APIServiceExport) error {
 	errs := []error{}
 
-	if err := r.ensureControllers(ctx, name, export); err != nil {
+	if err := r.ensureControllers(ctx, namespace, name, export); err != nil {
 		errs = append(errs, err)
 	}
 
@@ -90,22 +84,17 @@ func (r *reconciler) reconcile(ctx context.Context, name string, export *kubebin
 	return utilerrors.NewAggregate(errs)
 }
 
-func (r *reconciler) ensureControllers(ctx context.Context, name string, export *kubebindv1alpha2.APIServiceExport) error {
+func (r *reconciler) ensureControllers(ctx context.Context, namespace, name string, export *kubebindv1alpha2.APIServiceExport) error {
 	logger := klog.FromContext(ctx)
+	exportKey := contextstore.Key(namespace + "." + name) // Key for the export
 
 	if export == nil {
-		// stop dangling syncers on delete
-		r.lock.Lock()
-		defer r.lock.Unlock()
-
 		// Clean up any controllers associated with this export
-		for key, c := range r.syncContext {
-			if strings.HasSuffix(key, "."+name) {
-				logger.V(1).Info("Stopping APIServiceExport sync", "key", key, "reason", "APIServiceExport deleted")
-				c.cancel()
-				delete(r.syncContext, key)
-			}
+		deleted := r.syncStore.BulkDeletePrefixed(exportKey)
+		for _, k := range deleted {
+			logger.V(1).Info("Stopping APIServiceExport sync", "key", k.Key(), "reason", "NoAPIServiceExport")
 		}
+
 		return nil
 	}
 
@@ -116,14 +105,9 @@ func (r *reconciler) ensureControllers(ctx context.Context, name string, export
 	}
 	if binding == nil {
 		// Stop all controllers for this export
-		r.lock.Lock()
-		defer r.lock.Unlock()
-		for key, c := range r.syncContext {
-			if strings.HasSuffix(key, "."+export.Name) {
-				logger.V(1).Info("Stopping APIServiceExport sync", "key", key, "reason", "NoAPIServiceBinding")
-				c.cancel()
-				delete(r.syncContext, key)
-			}
+		deleted := r.syncStore.BulkDeletePrefixed(exportKey)
+		for _, k := range deleted {
+			logger.V(1).Info("Stopping APIServiceExport sync", "key", k.Key(), "reason", "NoAPIServiceBinding")
 		}
 		return nil
 	}
@@ -139,15 +123,12 @@ func (r *reconciler) ensureControllers(ctx context.Context, name string, export
 		schema, err := r.getRemoteBoundSchema(ctx, name)
 		if err != nil {
 			if errors.IsNotFound(err) {
-				// Stop the controller for this schema if it exists
-				r.lock.Lock()
-				key := name + "." + export.Name
-				if c, found := r.syncContext[key]; found {
-					logger.V(1).Info("Stopping APIServiceExport resource sync", "key", key, "reason", "BoundSchema not found")
-					c.cancel()
-					delete(r.syncContext, key)
+				// Stop the controller for this schema if it does not exists.
+				key := contextstore.NewKey(namespace, name, name)
+				deleted := r.syncStore.BulkDeletePrefixed(key)
+				for _, k := range deleted {
+					logger.V(1).Info("Stopping APIServiceExport sync", "key", k.Key(), "reason", "BoundSchema not found")
 				}
-				r.lock.Unlock()
 				continue
 			}
 			errs = append(errs, err)
@@ -159,50 +140,45 @@ func (r *reconciler) ensureControllers(ctx context.Context, name string, export
 			errs = append(errs, err)
 		}
 
-		processedSchemas[name] = true
+		processedSchemas[name] = true // This is only schemas names (suffix)
 		isClusterScoped = schema.Spec.Scope == apiextensionsv1.ClusterScoped || schema.Spec.InformerScope == kubebindv1alpha2.ClusterScope
 	}
 
 	// Ensure controller for permission claims
-	if err := r.ensureControllersForPermissionClaims(ctx, binding, isClusterScoped); err != nil {
+	if err := r.ensureControllersForPermissionClaims(ctx, export, binding, isClusterScoped); err != nil {
 		errs = append(errs, err)
 	}
 
-	// Stop controllers for schemas that are no longer referenced
-	r.lock.Lock()
-	for key, c := range r.syncContext {
-		parts := strings.Split(key, ".")
-		if len(parts) == 2 && parts[1] == export.Name {
-			schemaName := parts[0]
-			if !processedSchemas[schemaName] {
-				logger.V(1).Info("Stopping APIServiceExport resource sync", "key", key, "reason", "Schema no longer referenced")
-				c.cancel()
-				delete(r.syncContext, key)
-			}
+	// Stop controllers for schemas that are no longer referenced.
+	// This will be `exportNamespace.exportName.<schemaName>`
+	contexts := r.syncStore.ListPrefixed(exportKey)
+	for _, c := range contexts {
+		schemaName := strings.TrimPrefix(string(c.Key()), string(exportKey)+".") // schemaName will be processed schema names.
+		if !processedSchemas[schemaName] {
+			logger.V(1).Info("Stopping APIServiceExport resource sync", "key", c.Key(), "reason", "Schema no longer referenced")
+			r.syncStore.Delete(c.Key())
 		}
+
 	}
-	r.lock.Unlock()
 
 	return utilerrors.NewAggregate(errs)
 }
 
 func (r *reconciler) ensureControllerForSchema(ctx context.Context, export *kubebindv1alpha2.APIServiceExport, schema *kubebindv1alpha2.BoundSchema) error {
 	logger := klog.FromContext(ctx)
-	key := schema.Name + "." + export.Name
+	key := contextstore.NewKey(export.Namespace, export.Name, schema.Name)
 
-	r.lock.Lock()
-	c, found := r.syncContext[key]
+	c, found := r.syncStore.Get(key)
 	if found {
-		if c.generation == export.Generation {
-			r.lock.Unlock()
+		if c.Generation == export.Generation {
 			return nil // all as expected
 		}
 
 		logger.V(1).Info("Stopping APIServiceExport resource sync", "key", key, "reason", "GenerationChanged", "generation", schema.Generation)
-		c.cancel()
-		delete(r.syncContext, key)
+		r.syncStore.Delete(key)
 	}
-	r.lock.Unlock()
+
+	// At this point we dont have a controller for this schema.
 
 	// start a new syncer
 	var syncVersion string
@@ -307,26 +283,24 @@ func (r *reconciler) ensureControllerForSchema(ctx context.Context, export *kube
 		go statusCtrl.Start(ctxWithCancel, 1)
 	}()
 
-	r.lock.Lock()
-	defer r.lock.Unlock()
-	if c, found := r.syncContext[key]; found {
-		c.cancel()
-	}
-	r.syncContext[key] = syncContext{
-		generation: schema.Generation,
-		cancel:     cancel,
-	}
+	r.syncStore.Set(key, contextstore.SyncContext{
+		Generation: schema.Generation,
+		Cancel:     cancel,
+	})
 
 	return nil
 }
 
 func (r *reconciler) ensureControllersForPermissionClaims(
 	ctx context.Context,
+	export *kubebindv1alpha2.APIServiceExport,
 	binding *kubebindv1alpha2.APIServiceBinding,
 	isClusterScoped bool, // schema.Spec.Scope == apiextensionsv1.ClusterScoped || schema.Spec.InformerScope == kubebindv1alpha2.ClusterScope
 ) error {
 	logger := klog.FromContext(ctx)
 
+	//ctxWithCancel, cancel := context.WithCancel(ctx)
+
 	dynamicProviderClient := dynamicclient.NewForConfigOrDie(r.providerConfig)
 	dynamicConsumerClient := dynamicclient.NewForConfigOrDie(r.consumerConfig)
 

pkg/konnector/controllers/contextstore/contextstore.go

@@ -21,9 +21,8 @@ package contextstore
 // Context are stored at the APIServiceExport level.
 
 import (
+	"strings"
 	"sync"
-
-	kubebindv1alpha2 "github.com/kube-bind/kube-bind/sdk/apis/kubebind/v1alpha2"
 )
 
 type Key string
@@ -32,14 +31,16 @@ func (k Key) String() string {
 	return string(k)
 }
 
-func NewKey(export kubebindv1alpha2.APIServiceExport) Key {
-	return Key(export.Namespace + "/" + export.Name)
+func NewKey(exportNamespace, exportName string, suffix ...string) Key {
+	return Key(exportNamespace + "." + exportName + "." + strings.Join(suffix, "."))
 }
 
 type Store interface {
 	Get(key Key) (SyncContext, bool)
+	ListPrefixed(prefix Key) []SyncContext
 	Set(key Key, value SyncContext)
 	Delete(key Key)
+	BulkDeletePrefixed(prefix Key) []SyncContext
 }
 
 type contextStore struct {
@@ -48,8 +49,9 @@ type contextStore struct {
 }
 
 type SyncContext struct {
-	generation int64
-	cancel     func()
+	key        Key // reference key for the context for logging
+	Generation int64
+	Cancel     func()
 }
 
 func New() Store {
@@ -58,12 +60,8 @@ func New() Store {
 	}
 }
 
-func (c *SyncContext) Generation() int64 {
-	return c.generation
-}
-
-func (c *SyncContext) Cancel() {
-	c.cancel()
+func (c *SyncContext) Key() Key {
+	return c.key
 }
 
 func (c *contextStore) Get(key Key) (SyncContext, bool) {
@@ -73,9 +71,23 @@ func (c *contextStore) Get(key Key) (SyncContext, bool) {
 	return val, ok
 }
 
+func (c *contextStore) ListPrefixed(prefix Key) []SyncContext {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	var results []SyncContext
+	for k, v := range c.store {
+		if strings.HasPrefix(k.String(), prefix.String()) {
+			results = append(results, v)
+		}
+	}
+	return results
+}
+
 func (c *contextStore) Set(key Key, value SyncContext) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
+	value.key = key
 	c.store[key] = value
 }
 
@@ -84,7 +96,30 @@ func (c *contextStore) Delete(key Key) {
 	defer c.lock.Unlock()
 	ctx, ok := c.store[key]
 	if ok {
-		ctx.cancel()
+		ctx.Cancel()
 	}
 	delete(c.store, key)
 }
+
+func (c *contextStore) BulkDeletePrefixed(prefix Key) []SyncContext {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	var deleted []SyncContext
+	var keysToDelete []Key
+
+	for k, v := range c.store {
+		if strings.HasPrefix(k.String(), prefix.String()) {
+			keysToDelete = append(keysToDelete, k)
+			deleted = append(deleted, v)
+		}
+	}
+
+	for _, k := range keysToDelete {
+		ctx := c.store[k]
+		ctx.Cancel()
+		delete(c.store, k)
+	}
+
+	return deleted
+}