wire in pooling

Author: mjudeikis

v2/README.md

@@ -33,8 +33,12 @@ v2/
   (`schema.source: CRD`, single served version, no conversion webhook) onto the
   consumer, and report `Ready` + `boundAPIs`.
 - A dynamic per-GVR syncer copies instance **spec up** (server-side apply with
-  ownership markers + a finalizer) and **status down**, with `conflictPolicy:
-  Fail` surfaced as an Event (+ best-effort object condition).
+  ownership markers + a finalizer) and **status down**. `conflictPolicy: Fail`
+  refuses a foreign provider target (Event + conflict annotation, counted on the
+  binding's `conflictCount` + `Conflicts` condition); `conflictPolicy: Adopt`
+  takes over an *un-owned* provider object (never one owned by another binding).
+- The Connection **re-discovers** exported APIs periodically, so a CRD labeled
+  `exported` after connect is picked up and its binding goes Ready.
 - The provider side is the **multicluster-runtime engaged cluster** for each
   Connection: writes go through its client, fresh reads through its API reader,
   and status/drift events arrive via a **watch on its cache** (event-driven, not
@@ -50,8 +54,9 @@ v2/
   the provider — so `kubectl delete -f bundle.yaml` is order-don't-care.
 
 Known POC simplifications (tracked against the proposal): `schema.source:
-OpenAPI`, `conflictPolicy: Adopt`, related resources, `autoBind` and the
-provider `Lease` are not implemented yet.
+OpenAPI`, related resources, `pullPolicy: All`/`None`, `updatePolicy: Always`,
+`autoBind`, `deletion-policy: Orphan` and the provider `Lease` are not
+implemented yet.
 
 ## Build
 

v2/konnector/engine/binding/cleanup.go

@@ -84,6 +84,27 @@ func (b *base) cleanup(ctx context.Context, obj corev1alpha1.BindingAccessor, na
 	return nil
 }
 
+// countConflicts counts consumer instances of gvr in scope that the syncer
+// marked as conflicting (the conflict annotation).
+func (b *base) countConflicts(ctx context.Context, gvr schema.GroupVersionResource, namespace string) int32 {
+	ri := b.dyn.Resource(gvr)
+	var lister dynamicLister = ri
+	if namespace != "" {
+		lister = ri.Namespace(namespace)
+	}
+	list, err := lister.List(ctx, metav1.ListOptions{})
+	if err != nil {
+		return 0
+	}
+	var n int32
+	for i := range list.Items {
+		if list.Items[i].GetAnnotations()[corev1alpha1.AnnotationConflict] != "" {
+			n++
+		}
+	}
+	return n
+}
+
 // drainInstances deletes provider copies and releases the syncer finalizer on
 // every consumer instance of gvr in scope.
 func (b *base) drainInstances(ctx context.Context, gvr schema.GroupVersionResource, namespace string, providerClient client.Client, localUID string) error {

v2/konnector/engine/binding/reconciler.go

@@ -26,6 +26,7 @@ import (
 	"fmt"
 	"sort"
 	"strings"
+	"time"
 
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -49,9 +50,19 @@ type base struct {
 	client    client.Client
 	dyn       dynamic.Interface
 	scheme    *runtime.Scheme
+	resync    time.Duration
 	newClient func(ctx context.Context, conn *corev1alpha1.Connection) (client.Client, error)
 }
 
+// resyncInterval is how often a Ready binding re-reconciles to refresh
+// conflictCount (conflicts are detected by the syncer, not via binding events).
+func (b *base) resyncInterval() time.Duration {
+	if b.resync > 0 {
+		return b.resync
+	}
+	return 30 * time.Second
+}
+
 func (b *base) providerClient(ctx context.Context, conn *corev1alpha1.Connection) (client.Client, error) {
 	if b.newClient != nil {
 		return b.newClient(ctx, conn)
@@ -60,7 +71,9 @@ func (b *base) providerClient(ctx context.Context, conn *corev1alpha1.Connection
 }
 
 // reconcileAccessor drives a binding (cluster or namespaced) toward Ready.
-func (b *base) reconcileAccessor(ctx context.Context, obj corev1alpha1.BindingAccessor) error {
+// namespace scopes conflict counting: "" for a ClusterBinding (cluster-wide),
+// the Binding's namespace otherwise.
+func (b *base) reconcileAccessor(ctx context.Context, obj corev1alpha1.BindingAccessor, namespace string) error {
 	spec := obj.BindingSpecP()
 	status := obj.BindingStatusP()
 
@@ -99,7 +112,12 @@ func (b *base) reconcileAccessor(ctx context.Context, obj corev1alpha1.BindingAc
 		if err != nil {
 			return fmt.Errorf("pulling CRD %q: %w", api.Name, err)
 		}
-		boundAPIs = append(boundAPIs, corev1alpha1.BoundAPI{Name: exported.Name, CRDHash: hash})
+		// Count instances we refused to sync due to a foreign provider target.
+		var conflicts int32
+		if gvr, ok, gerr := b.gvrFor(ctx, api.Name); gerr == nil && ok {
+			conflicts = b.countConflicts(ctx, gvr, namespace)
+		}
+		boundAPIs = append(boundAPIs, corev1alpha1.BoundAPI{Name: exported.Name, CRDHash: hash, ConflictCount: conflicts})
 	}
 	sort.Slice(boundAPIs, func(i, j int) bool { return boundAPIs[i].Name < boundAPIs[j].Name })
 	status.BoundAPIs = boundAPIs
@@ -110,6 +128,17 @@ func (b *base) reconcileAccessor(ctx context.Context, obj corev1alpha1.BindingAc
 		return nil
 	}
 
+	var totalConflicts int32
+	for _, ba := range boundAPIs {
+		totalConflicts += ba.ConflictCount
+	}
+	if totalConflicts > 0 {
+		apimeta.SetStatusCondition(&status.Conditions, condition(obj, corev1alpha1.ConditionConflicts, metav1.ConditionTrue, corev1alpha1.ReasonForeignObjectExists,
+			fmt.Sprintf("%d object(s) skipped due to foreign ownership; see object Events", totalConflicts)))
+	} else {
+		apimeta.SetStatusCondition(&status.Conditions, condition(obj, corev1alpha1.ConditionConflicts, metav1.ConditionFalse, corev1alpha1.ReasonAsExpected, "no conflicts"))
+	}
+
 	apimeta.SetStatusCondition(&status.Conditions, condition(obj, corev1alpha1.ConditionSynced, metav1.ConditionTrue, corev1alpha1.ReasonAsExpected, "all APIs installed"))
 	setReady(status, obj, metav1.ConditionTrue, corev1alpha1.ReasonAsExpected, "binding ready")
 	return nil
@@ -209,14 +238,20 @@ func setReady(status *corev1alpha1.BindingStatus, obj client.Object, s metav1.Co
 // ----- ClusterBinding -----
 
 // ClusterReconciler reconciles ClusterBinding objects.
-type ClusterReconciler struct{ base }
+type ClusterReconciler struct {
+	base
+	// Resync is how often a Ready binding re-reconciles to refresh conflictCount
+	// (0 = default 30s).
+	Resync time.Duration
+}
 
 // SetupWithManager registers the ClusterBinding reconciler. It also watches
 // Connection so bindings re-reconcile when their connection becomes Ready or
 // its exported APIs change (level-triggered).
 func (r *ClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	r.base.client = mgr.GetClient()
 	r.base.scheme = mgr.GetScheme()
+	r.base.resync = r.Resync
 	dyn, err := dynamic.NewForConfig(mgr.GetConfig())
 	if err != nil {
 		return err
@@ -264,25 +299,31 @@ func (r *ClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		return ctrl.Result{}, r.base.client.Update(ctx, cb)
 	}
 	orig := cb.DeepCopy()
-	rerr := r.base.reconcileAccessor(ctx, cb)
+	rerr := r.base.reconcileAccessor(ctx, cb, "")
 	if !equalBindingStatus(&orig.Status, &cb.Status) {
 		if err := r.base.client.Status().Update(ctx, cb); err != nil {
 			return ctrl.Result{}, err
 		}
 	}
-	return ctrl.Result{}, rerr
+	return ctrl.Result{RequeueAfter: r.base.resyncInterval()}, rerr
 }
 
 // ----- Binding (namespaced) -----
 
 // NamespacedReconciler reconciles Binding objects.
-type NamespacedReconciler struct{ base }
+type NamespacedReconciler struct {
+	base
+	// Resync is how often a Ready binding re-reconciles to refresh conflictCount
+	// (0 = default 30s).
+	Resync time.Duration
+}
 
 // SetupWithManager registers the Binding reconciler. It also watches Connection
 // so bindings re-reconcile when their connection becomes Ready (level-triggered).
 func (r *NamespacedReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	r.base.client = mgr.GetClient()
 	r.base.scheme = mgr.GetScheme()
+	r.base.resync = r.Resync
 	dyn, err := dynamic.NewForConfig(mgr.GetConfig())
 	if err != nil {
 		return err
@@ -331,13 +372,13 @@ func (r *NamespacedReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		return ctrl.Result{}, r.base.client.Update(ctx, b)
 	}
 	orig := b.DeepCopy()
-	rerr := r.base.reconcileAccessor(ctx, b)
+	rerr := r.base.reconcileAccessor(ctx, b, b.Namespace)
 	if !equalBindingStatus(&orig.Status, &b.Status) {
 		if err := r.base.client.Status().Update(ctx, b); err != nil {
 			return ctrl.Result{}, err
 		}
 	}
-	return ctrl.Result{}, rerr
+	return ctrl.Result{RequeueAfter: r.base.resyncInterval()}, rerr
 }
 
 func equalBindingStatus(a, b *corev1alpha1.BindingStatus) bool {

v2/konnector/engine/connection/reconciler.go

@@ -52,6 +52,16 @@ type Reconciler struct {
 	// Connection. Overridable in tests; defaults to a fresh client from the
 	// resolved kubeconfig.
 	NewProviderClient func(ctx context.Context, conn *corev1alpha1.Connection) (client.Client, error)
+	// DiscoveryResync is how often a Ready Connection re-discovers exported APIs
+	// (so a CRD labeled exported after connect is picked up). 0 = default 30s.
+	DiscoveryResync time.Duration
+}
+
+func (r *Reconciler) discoveryResync() time.Duration {
+	if r.DiscoveryResync > 0 {
+		return r.DiscoveryResync
+	}
+	return 30 * time.Second
 }
 
 // SetupWithManager registers the reconciler with the consumer manager. It also
@@ -126,7 +136,13 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 			return ctrl.Result{}, err
 		}
 	}
-	return ctrl.Result{}, reconcileErr
+	// Periodically re-discover so a CRD labeled exported after connect is picked
+	// up (the binding watches the Connection and reacts to exportedAPIs changes).
+	result := ctrl.Result{}
+	if reconcileErr == nil {
+		result.RequeueAfter = r.discoveryResync()
+	}
+	return result, reconcileErr
 }
 
 func (r *Reconciler) reconcile(ctx context.Context, conn *corev1alpha1.Connection) error {

v2/konnector/engine/sync/resolve.go

@@ -39,6 +39,8 @@ type Resolution struct {
 	Ready bool
 	// ConnectionName is the Connection the binding references.
 	ConnectionName string
+	// ConflictPolicy is the binding's conflict policy (Fail if unset).
+	ConflictPolicy corev1alpha1.ConflictPolicy
 }
 
 // ResolveConnection finds the binding that covers crdName for the given
@@ -61,6 +63,7 @@ func ResolveConnection(ctx context.Context, c client.Client, crdName, namespace
 				Found:          true,
 				Ready:          apimeta.IsStatusConditionTrue(cb.Status.Conditions, corev1alpha1.ConditionReady),
 				ConnectionName: cb.Spec.ConnectionRef.Name,
+				ConflictPolicy: cb.Spec.ConflictPolicy,
 			}, nil
 		}
 	}
@@ -80,6 +83,7 @@ func ResolveConnection(ctx context.Context, c client.Client, crdName, namespace
 					Found:          true,
 					Ready:          apimeta.IsStatusConditionTrue(b.Status.Conditions, corev1alpha1.ConditionReady),
 					ConnectionName: b.Spec.ConnectionRef.Name,
+					ConflictPolicy: b.Spec.ConflictPolicy,
 				}, nil
 			}
 		}

v2/konnector/engine/sync/syncer.go

@@ -27,6 +27,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/dynamic/dynamiclister"
 	"k8s.io/client-go/tools/record"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -128,17 +129,18 @@ func (r *specReconciler) Reconcile(ctx context.Context, req reconcile.Request) (
 	switch {
 	case getErr == nil:
 		if owner, ours := ownershipOf(existing, r.localUID, string(obj.GetUID())); !ours {
-			reason := conflictReason(owner)
-			msg := fmt.Sprintf("provider object %s already exists and is %s; not overwriting (conflictPolicy: Fail)", client.ObjectKeyFromObject(obj), owner)
-			log.Info("conflict: refusing to overwrite provider object", "owner", owner)
-			// Surface on the object's own condition (best-effort: pruned if the
-			// CRD's status schema has no conditions) AND as an Event (always).
-			setObjCondition(obj, metav1.ConditionFalse, reason, msg)
-			_ = r.consumerClient.Status().Update(ctx, obj)
-			if r.recorder != nil {
-				r.recorder.Event(obj, corev1.EventTypeWarning, reason, msg)
+			// Adopt only takes an un-owned (markerless) object; it never steals
+			// one already owned by another binding/consumer.
+			if owner != ownerForeignUnmanaged || res.ConflictPolicy != corev1alpha1.ConflictPolicyAdopt {
+				reason := conflictReason(owner)
+				msg := fmt.Sprintf("provider object %s already exists and is %s; not overwriting (conflictPolicy: %s)", client.ObjectKeyFromObject(obj), owner, policyOrFail(res.ConflictPolicy))
+				log.Info("conflict: refusing to overwrite provider object", "owner", owner)
+				if err := r.markConflict(ctx, obj, reason, msg); err != nil {
+					return reconcile.Result{}, err
+				}
+				return reconcile.Result{}, nil
 			}
-			return reconcile.Result{}, nil
+			log.Info("adopting un-owned provider object", "key", client.ObjectKeyFromObject(obj))
 		}
 	case apierrors.IsNotFound(getErr):
 		if r.scope == apiextensionsv1.NamespaceScoped {
@@ -150,6 +152,11 @@ func (r *specReconciler) Reconcile(ctx context.Context, req reconcile.Request) (
 		return reconcile.Result{}, fmt.Errorf("provider get: %w", getErr)
 	}
 
+	// We are syncing — clear any stale conflict marker from a previous round.
+	if err := r.clearConflict(ctx, obj); err != nil {
+		return reconcile.Result{}, err
+	}
+
 	// Spec consumer -> provider (SSA).
 	patch := r.providerPatch(obj, r.localUID)
 	if err := r.providerClient.Patch(ctx, patch, client.Apply, client.FieldOwner(fieldOwner), client.ForceOwnership); err != nil {
@@ -168,6 +175,36 @@ func (r *specReconciler) Reconcile(ctx context.Context, req reconcile.Request) (
 	return reconcile.Result{RequeueAfter: statusResyncBackstop}, nil
 }
 
+// markConflict records a conflict durably (an annotation that survives status
+// pruning, used by bindings to count conflicts), best-effort on the object's
+// own condition, and as an Event.
+func (r *specReconciler) markConflict(ctx context.Context, obj *unstructured.Unstructured, reason, msg string) error {
+	if obj.GetAnnotations()[corev1alpha1.AnnotationConflict] != reason {
+		p := fmt.Appendf(nil, `{"metadata":{"annotations":{%q:%q}}}`, corev1alpha1.AnnotationConflict, reason)
+		if err := r.consumerClient.Patch(ctx, obj, client.RawPatch(types.MergePatchType, p)); client.IgnoreNotFound(err) != nil {
+			return fmt.Errorf("marking conflict: %w", err)
+		}
+	}
+	setObjCondition(obj, metav1.ConditionFalse, reason, msg)
+	_ = r.consumerClient.Status().Update(ctx, obj)
+	if r.recorder != nil {
+		r.recorder.Event(obj, corev1.EventTypeWarning, reason, msg)
+	}
+	return nil
+}
+
+// clearConflict removes the conflict marker once the object syncs cleanly.
+func (r *specReconciler) clearConflict(ctx context.Context, obj *unstructured.Unstructured) error {
+	if obj.GetAnnotations()[corev1alpha1.AnnotationConflict] == "" {
+		return nil
+	}
+	p := fmt.Appendf(nil, `{"metadata":{"annotations":{%q:null}}}`, corev1alpha1.AnnotationConflict)
+	if err := r.consumerClient.Patch(ctx, obj, client.RawPatch(types.MergePatchType, p)); client.IgnoreNotFound(err) != nil {
+		return fmt.Errorf("clearing conflict: %w", err)
+	}
+	return nil
+}
+
 func (r *specReconciler) reconcileDelete(ctx context.Context, obj *unstructured.Unstructured) (reconcile.Result, error) {
 	if !controllerutil.ContainsFinalizer(obj, corev1alpha1.FinalizerSyncer) {
 		return reconcile.Result{}, nil
@@ -241,28 +278,42 @@ func newUnstructured(gvk schema.GroupVersionKind) *unstructured.Unstructured {
 	return u
 }
 
+// Ownership classifications returned by ownershipOf.
+const (
+	ownerForeignUnmanaged = "foreign-unmanaged"
+	ownerOurs             = "ours"
+	ownerByAnother        = "owned-by-another"
+)
+
 // ownershipOf reports the marker owner of a provider object and whether it is
 // ours (our consumer cluster + our consumer object UID).
 func ownershipOf(obj *unstructured.Unstructured, localUID, consumerObjUID string) (string, bool) {
 	ann := obj.GetAnnotations()
 	cluster := ann[corev1alpha1.AnnotationConsumerClusterUID]
 	objUID := ann[corev1alpha1.AnnotationConsumerObjectUID]
 	if cluster == "" && objUID == "" {
-		return "foreign-unmanaged", false
+		return ownerForeignUnmanaged, false
 	}
 	if cluster == localUID && objUID == consumerObjUID {
-		return "ours", true
+		return ownerOurs, true
 	}
-	return "owned-by-another", false
+	return ownerByAnother, false
 }
 
 func conflictReason(owner string) string {
-	if owner == "foreign-unmanaged" {
+	if owner == ownerForeignUnmanaged {
 		return corev1alpha1.ReasonForeignObjectExists
 	}
 	return corev1alpha1.ReasonOwnedByAnother
 }
 
+func policyOrFail(p corev1alpha1.ConflictPolicy) corev1alpha1.ConflictPolicy {
+	if p == "" {
+		return corev1alpha1.ConflictPolicyFail
+	}
+	return p
+}
+
 func setObjCondition(obj *unstructured.Unstructured, status metav1.ConditionStatus, reason, msg string) {
 	cond := map[string]any{
 		"type":    corev1alpha1.ConditionSynced,