Skip to content

Move to schemaless backend #295

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mjudeikis merged 3 commits into from
Sep 18, 2025
Merged

Move to schemaless backend #295

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
1b19d69
Remove APIResourceSchemas
mjudeikis Sep 10, 2025
3b83bce
Address review comments.
mjudeikis Sep 12, 2025
b02f846
Address more comments
mjudeikis Sep 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions backend/controllers/clusterbinding/clusterbinding_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,11 +77,11 @@ func NewClusterBindingReconciler(
}
return exports, nil
},
getAPIResourceSchema: func(ctx context.Context, cache cache.Cache, name string) (*kubebindv1alpha2.APIResourceSchema, error) {
result := &kubebindv1alpha2.APIResourceSchema{}
err := cache.Get(ctx, types.NamespacedName{Name: name}, result)
getBoundSchema: func(ctx context.Context, cache cache.Cache, namespace, name string) (*kubebindv1alpha2.BoundSchema, error) {
result := &kubebindv1alpha2.BoundSchema{}
err := cache.Get(ctx, types.NamespacedName{Namespace: namespace, Name: name}, result)
if err != nil {
return nil, fmt.Errorf("failed to get APIResourceSchema %q: %w", name, err)
return nil, fmt.Errorf("failed to get BoundSchema %q: %w", name, err)
}
return result, nil
},
Expand Down
48 changes: 30 additions & 18 deletions backend/controllers/clusterbinding/clusterbinding_reconcile.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,9 @@ package clusterbinding
import (
"context"
"fmt"
"maps"
"reflect"
"slices"
"time"

corev1 "k8s.io/api/core/v1"
Expand All @@ -40,11 +42,11 @@ import (
type reconciler struct {
scope kubebindv1alpha2.InformerScope

listServiceExports func(ctx context.Context, cache cache.Cache, ns string) ([]*kubebindv1alpha2.APIServiceExport, error)
getAPIResourceSchema func(ctx context.Context, cache cache.Cache, name string) (*kubebindv1alpha2.APIResourceSchema, error)
getClusterRole func(ctx context.Context, cache cache.Cache, name string) (*rbacv1.ClusterRole, error)
createClusterRole func(ctx context.Context, client client.Client, binding *rbacv1.ClusterRole) error
updateClusterRole func(ctx context.Context, client client.Client, binding *rbacv1.ClusterRole) error
listServiceExports func(ctx context.Context, cache cache.Cache, ns string) ([]*kubebindv1alpha2.APIServiceExport, error)
getBoundSchema func(ctx context.Context, cache cache.Cache, namespace, name string) (*kubebindv1alpha2.BoundSchema, error)
getClusterRole func(ctx context.Context, cache cache.Cache, name string) (*rbacv1.ClusterRole, error)
createClusterRole func(ctx context.Context, client client.Client, binding *rbacv1.ClusterRole) error
updateClusterRole func(ctx context.Context, client client.Client, binding *rbacv1.ClusterRole) error

getClusterRoleBinding func(ctx context.Context, cache cache.Cache, name string) (*rbacv1.ClusterRoleBinding, error)
createClusterRoleBinding func(ctx context.Context, client client.Client, binding *rbacv1.ClusterRoleBinding) error
Expand Down Expand Up @@ -148,24 +150,35 @@ func (r *reconciler) ensureRBACClusterRole(ctx context.Context, client client.Cl
},
},
},
}
Rules: []rbacv1.PolicyRule{
// Always need to be able to get/list/watch the BoundSchemas
// to be able to figure out what to bind.
{
APIGroups: []string{kubebindv1alpha2.GroupName},
Resources: []string{"boundschemas"},
Verbs: []string{"get", "list", "watch"},
},
}}
for _, export := range exports {
// Collect unique GroupResources and sort for stable rule ordering.
grSet := map[string]kubebindv1alpha2.GroupResource{}
for _, res := range export.Spec.Resources {
schema, err := r.getAPIResourceSchema(ctx, cache, res.Name)
key := res.ResourceGroupName()
grSet[key] = kubebindv1alpha2.GroupResource{Group: res.Group, Resource: res.Resource}
}
keys := slices.Collect(maps.Keys(grSet))
slices.Sort(keys)
for _, k := range keys {
// k is already normalized (e.g., "pods.core" for empty group).
Comment thread
mjudeikis marked this conversation as resolved.
schema, err := r.getBoundSchema(ctx, cache, clusterBinding.Namespace, k)
if err != nil {
return fmt.Errorf("failed to get APIResourceSchema %w", err)
return fmt.Errorf("failed to get BoundSchema %q: %w", k, err)
}

expected.Rules = append(expected.Rules,
rbacv1.PolicyRule{
APIGroups: []string{schema.Spec.APIResourceSchemaCRDSpec.Group},
Resources: []string{schema.Spec.APIResourceSchemaCRDSpec.Names.Plural},
Verbs: []string{"get", "list", "watch", "update", "patch", "delete", "create"},
},
rbacv1.PolicyRule{
APIGroups: []string{kubebindv1alpha2.GroupName},
Resources: []string{"apiresourceschemas"},
Verbs: []string{"get", "list", "watch"},
APIGroups: []string{schema.Spec.Group},
Resources: []string{schema.Spec.Names.Plural},
Verbs: []string{"get", "list", "watch", "create", "update", "patch", "delete"},
},
)
}
Expand All @@ -192,7 +205,6 @@ func (r *reconciler) ensureRBACClusterRoleBinding(ctx context.Context, client cl
if err != nil && !errors.IsNotFound(err) {
return fmt.Errorf("failed to get ClusterRoleBinding %s: %w", name, err)
}

if r.scope != kubebindv1alpha2.ClusterScope {
if err := r.deleteClusterRoleBinding(ctx, client, name); err != nil && !errors.IsNotFound(err) {
return fmt.Errorf("failed to delete ClusterRoleBinding %s: %w", name, err)
Expand Down
42 changes: 26 additions & 16 deletions backend/controllers/serviceexport/serviceexport_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,18 +58,18 @@ func NewAPIServiceExportReconciler(
mgr mcmanager.Manager,
opts controller.TypedOptions[mcreconcile.Request],
) (*APIServiceExportReconciler, error) {
if err := mgr.GetFieldIndexer().IndexField(ctx, &kubebindv1alpha2.APIServiceExport{}, indexers.ServiceExportByAPIResourceSchema,
indexers.IndexServiceExportByAPIResourceSchema); err != nil {
return nil, fmt.Errorf("failed to setup ServiceExportByAPIResourceSchema indexer: %w", err)
if err := mgr.GetFieldIndexer().IndexField(ctx, &kubebindv1alpha2.APIServiceExport{}, indexers.ServiceExportByBoundSchema,
indexers.IndexServiceExportByBoundSchema); err != nil {
return nil, fmt.Errorf("failed to setup ServiceExportByBoundSchema indexer: %w", err)
}

r := &APIServiceExportReconciler{
manager: mgr,
opts: opts,
reconciler: reconciler{
getAPIResourceSchema: func(ctx context.Context, cache cache.Cache, name string) (*kubebindv1alpha2.APIResourceSchema, error) {
var schema kubebindv1alpha2.APIResourceSchema
key := types.NamespacedName{Name: name}
getBoundSchema: func(ctx context.Context, cache cache.Cache, namespace, name string) (*kubebindv1alpha2.BoundSchema, error) {
var schema kubebindv1alpha2.BoundSchema
key := types.NamespacedName{Namespace: namespace, Name: name}
if err := cache.Get(ctx, key, &schema); err != nil {
return nil, err
}
Expand All @@ -92,6 +92,8 @@ func NewAPIServiceExportReconciler(
//+kubebuilder:rbac:groups=kubebind.k8s.io,resources=apiserviceexports,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=kubebind.k8s.io,resources=apiserviceexports/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=kubebind.k8s.io,resources=apiserviceexports/finalizers,verbs=update
//+kubebuilder:rbac:groups=kubebind.k8s.io,resources=boundschemas,verbs=get;list;watch
//+kubebuilder:rbac:groups=kubebind.k8s.io,resources=boundschemas/status,verbs=get;update;patch

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
Expand Down Expand Up @@ -128,10 +130,18 @@ func (r *APIServiceExportReconciler) Reconcile(ctx context.Context, req mcreconc
return ctrl.Result{}, err
}

// Update status if it has changed
if !equality.Semantic.DeepEqual(original, apiServiceExport) {
err := client.Status().Update(ctx, apiServiceExport)
if err != nil {
// Update annotations changed (hash), we need to propagate it and requeue for status changes.
// This is why we compare annotations only as we don't expect any changes to spec.
// Status changes are handled below.
if !equality.Semantic.DeepEqual(original.Annotations, apiServiceExport.Annotations) {
if err := client.Update(ctx, apiServiceExport); err != nil {
return ctrl.Result{}, fmt.Errorf("failed to update APIServiceExport: %w", err)
}
logger.Info("APIServiceExport hash updated", "namespace", apiServiceExport.Namespace, "name", apiServiceExport.Name)
return ctrl.Result{Requeue: true}, nil
}
if !equality.Semantic.DeepEqual(original.Status, apiServiceExport.Status) {
if err := client.Status().Update(ctx, apiServiceExport); err != nil {
return ctrl.Result{}, fmt.Errorf("failed to update APIServiceExport status: %w", err)
}
logger.Info("APIServiceExport status updated", "namespace", apiServiceExport.Namespace, "name", apiServiceExport.Name)
Expand All @@ -141,14 +151,14 @@ func (r *APIServiceExportReconciler) Reconcile(ctx context.Context, req mcreconc
}

// getAPIResourceSchemaMapper returns a mapper function that uses the manager to find related APIServiceExports.
func getAPIResourceSchemaMapper(clusterName string, cl cluster.Cluster) handler.TypedEventHandler[client.Object, mcreconcile.Request] {
func getBoundSchemaMapper(clusterName string, cl cluster.Cluster) handler.TypedEventHandler[client.Object, mcreconcile.Request] {
return handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, obj client.Object) []mcreconcile.Request {
apiResourceSchema := obj.(*kubebindv1alpha2.APIResourceSchema)
apiResourceSchemaKey := apiResourceSchema.Name
boundSchema := obj.(*kubebindv1alpha2.BoundSchema)
boundSchemaKey := boundSchema.Spec.Names.Plural + "." + boundSchema.Spec.Group
c := cl.GetClient()

var exports kubebindv1alpha2.APIServiceExportList
if err := c.List(ctx, &exports, client.MatchingFields{indexers.ServiceExportByAPIResourceSchema: apiResourceSchemaKey}); err != nil {
if err := c.List(ctx, &exports, client.MatchingFields{indexers.ServiceExportByBoundSchema: boundSchemaKey}); err != nil {
return []mcreconcile.Request{}
}

Expand All @@ -171,8 +181,8 @@ func (r *APIServiceExportReconciler) SetupWithManager(mgr mcmanager.Manager) err
return mcbuilder.ControllerManagedBy(mgr).
For(&kubebindv1alpha2.APIServiceExport{}).
Watches(
&kubebindv1alpha2.APIResourceSchema{},
getAPIResourceSchemaMapper,
&kubebindv1alpha2.BoundSchema{},
getBoundSchemaMapper,
).
Comment thread
mjudeikis marked this conversation as resolved.
WithOptions(r.opts).
Named(controllerName).
Expand Down
70 changes: 28 additions & 42 deletions backend/controllers/serviceexport/serviceexport_reconcile.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,6 @@ package serviceexport

import (
"context"
"crypto/sha256"
"encoding/hex"
"slices"
"sort"

"k8s.io/apimachinery/pkg/api/errors"
utilerrors "k8s.io/apimachinery/pkg/util/errors"
Expand All @@ -30,75 +26,65 @@ import (
"sigs.k8s.io/controller-runtime/pkg/client"

kubebindv1alpha2 "github.com/kube-bind/kube-bind/sdk/apis/kubebind/v1alpha2"
kubebindhelpers "github.com/kube-bind/kube-bind/sdk/apis/kubebind/v1alpha2/helpers"
"github.com/kube-bind/kube-bind/sdk/apis/kubebind/v1alpha2/helpers"
conditionsapi "github.com/kube-bind/kube-bind/sdk/apis/third_party/conditions/apis/conditions/v1alpha1"
"github.com/kube-bind/kube-bind/sdk/apis/third_party/conditions/util/conditions"
)

type reconciler struct {
getAPIResourceSchema func(ctx context.Context, cache cache.Cache, name string) (*kubebindv1alpha2.APIResourceSchema, error)
deleteServiceExport func(ctx context.Context, client client.Client, namespace, name string) error
getBoundSchema func(ctx context.Context, cache cache.Cache, namespace, name string) (*kubebindv1alpha2.BoundSchema, error)
deleteServiceExport func(ctx context.Context, client client.Client, namespace, name string) error
}

func (r *reconciler) reconcile(ctx context.Context, cache cache.Cache, export *kubebindv1alpha2.APIServiceExport) error {
var errs []error

if specChanged, err := r.ensureSchema(ctx, cache, export); err != nil {
if err := r.ensureSchema(ctx, cache, export); err != nil {
errs = append(errs, err)
} else if specChanged {
// TODO: This should be separate controller for apiresourceschemas.
// This is wrong place now.
// r.requeue(export)
return nil
}

return utilerrors.NewAggregate(errs)
}

func (r *reconciler) ensureSchema(ctx context.Context, cache cache.Cache, export *kubebindv1alpha2.APIServiceExport) (specChanged bool, err error) {
func (r *reconciler) ensureSchema(ctx context.Context, cache cache.Cache, export *kubebindv1alpha2.APIServiceExport) error {
logger := klog.FromContext(ctx)
leafHashes := make([]string, 0, len(export.Spec.Resources))
for _, resourceRef := range export.Spec.Resources {
if resourceRef.Type != "APIResourceSchema" {
logger.V(1).Info("Skipping unsupported resource type", "type", resourceRef.Type)
continue
}
schemas := make([]*kubebindv1alpha2.BoundSchema, 0, len(export.Spec.Resources))

schema, err := r.getAPIResourceSchema(ctx, cache, resourceRef.Name)
for _, res := range export.Spec.Resources {
name := res.ResourceGroupName()
schema, err := r.getBoundSchema(ctx, cache, export.Namespace, name)
if err != nil {
if errors.IsNotFound(err) {
continue
conditions.MarkFalse(
export,
kubebindv1alpha2.APIServiceExportConditionProviderInSync,
"BoundSchemaMissing",
conditionsapi.ConditionSeverityError,
"BoundSchema %q is not available: %v", name, err)
return nil
}
return false, err
return err
}

hash := kubebindhelpers.APIResourceSchemaCRDSpecHash(&schema.Spec.APIResourceSchemaCRDSpec)
leafHashes = append(leafHashes, hash)
schemas = append(schemas, schema)
}

hashOfHashes := hashOfHashes(leafHashes)
hash, err := helpers.BoundSchemasSpecHash(schemas)
if err != nil {
return err
}
Comment thread
mjudeikis marked this conversation as resolved.
Comment thread
mjudeikis marked this conversation as resolved.

if export.Annotations[kubebindv1alpha2.SourceSpecHashAnnotationKey] != hashOfHashes {
if export.Annotations[kubebindv1alpha2.SourceSpecHashAnnotationKey] != hash {
// both exist, update APIServiceExport
logger.V(1).Info("Updating APIServiceExport. Hash mismatch", "hash", hashOfHashes, "expected", export.Annotations[kubebindv1alpha2.SourceSpecHashAnnotationKey])
logger.V(1).Info("Updating APIServiceExport. Hash mismatch", "hash", hash, "expected", export.Annotations[kubebindv1alpha2.SourceSpecHashAnnotationKey])
if export.Annotations == nil {
export.Annotations = map[string]string{}
}
export.Annotations[kubebindv1alpha2.SourceSpecHashAnnotationKey] = hashOfHashes
return true, nil
export.Annotations[kubebindv1alpha2.SourceSpecHashAnnotationKey] = hash
return nil
}

conditions.MarkTrue(export, kubebindv1alpha2.APIServiceExportConditionProviderInSync)

return false, nil
}

func hashOfHashes(hashes []string) string {
hexHashes := slices.Clone(hashes)
sort.Strings(hexHashes)

rootHasher := sha256.New()
for _, h := range hexHashes {
rootHasher.Write([]byte(h))
}
return hex.EncodeToString(rootHasher.Sum(nil))
return nil
}
40 changes: 23 additions & 17 deletions backend/controllers/serviceexportrequest/serviceexportrequest_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ func NewAPIServiceExportRequestReconciler(
opts controller.TypedOptions[mcreconcile.Request],
scope kubebindv1alpha2.InformerScope,
isolation kubebindv1alpha2.Isolation,
schemaSource string,
) (*APIServiceExportRequestReconciler, error) {
Comment thread
mjudeikis marked this conversation as resolved.
// Set up field indexers for APIServiceExportRequests
if err := mgr.GetFieldIndexer().IndexField(ctx, &kubebindv1alpha2.APIServiceExportRequest{}, indexers.ServiceExportRequestByServiceExport,
Expand All @@ -81,9 +82,10 @@ func NewAPIServiceExportRequestReconciler(
reconciler: reconciler{
informerScope: scope,
clusterScopedIsolation: isolation,
getAPIResourceSchema: func(ctx context.Context, cache cache.Cache, name string) (*kubebindv1alpha2.APIResourceSchema, error) {
var schema kubebindv1alpha2.APIResourceSchema
key := types.NamespacedName{Name: name}
schemaSource: schemaSource,
getBoundSchema: func(ctx context.Context, cache cache.Cache, namespace, name string) (*kubebindv1alpha2.BoundSchema, error) {
var schema kubebindv1alpha2.BoundSchema
key := types.NamespacedName{Namespace: namespace, Name: name}
if err := cache.Get(ctx, key, &schema); err != nil {
return nil, err
}
Expand All @@ -100,11 +102,8 @@ func NewAPIServiceExportRequestReconciler(
createServiceExport: func(ctx context.Context, cl client.Client, resource *kubebindv1alpha2.APIServiceExport) error {
return cl.Create(ctx, resource)
},
createAPIResourceSchema: func(ctx context.Context, cl client.Client, schema *kubebindv1alpha2.APIResourceSchema) (*kubebindv1alpha2.APIResourceSchema, error) {
if err := cl.Create(ctx, schema); err != nil {
return nil, err
}
return schema, nil
createBoundSchema: func(ctx context.Context, cl client.Client, schema *kubebindv1alpha2.BoundSchema) error {
return cl.Create(ctx, schema)
},
Comment thread
mjudeikis marked this conversation as resolved.
deleteServiceExportRequest: func(ctx context.Context, cl client.Client, ns, name string) error {
return cl.Delete(ctx, &kubebindv1alpha2.APIServiceExportRequest{
Expand Down Expand Up @@ -150,15 +149,15 @@ func getServiceExportRequestMapper(clusterName string, cl cluster.Cluster) handl
})
}

// getAPIResourceSchemaMapper creates a mapping function for APIResourceSchema changes.
func getAPIResourceSchemaMapper(clusterName string, cl cluster.Cluster) handler.TypedEventHandler[client.Object, mcreconcile.Request] {
// getBoundSchemaMapper creates a mapping function for BoundSchema changes.
func getBoundSchemaMapper(clusterName string, cl cluster.Cluster) handler.TypedEventHandler[client.Object, mcreconcile.Request] {
return handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, obj client.Object) []mcreconcile.Request {
apiResourceSchema := obj.(*kubebindv1alpha2.APIResourceSchema)
apiResourceSchemaKey := apiResourceSchema.Name
boundSchema := obj.(*kubebindv1alpha2.BoundSchema)
boundSchemaKey := boundSchema.Spec.Names.Plural + "." + boundSchema.Spec.Group
c := cl.GetClient()

var requests kubebindv1alpha2.APIServiceExportRequestList
if err := c.List(ctx, &requests, client.MatchingFields{indexers.ServiceExportRequestByGroupResource: apiResourceSchemaKey}); err != nil {
if err := c.List(ctx, &requests, client.MatchingFields{indexers.ServiceExportRequestByGroupResource: boundSchemaKey}); err != nil {
return []mcreconcile.Request{}
}

Expand Down Expand Up @@ -216,13 +215,20 @@ func (r *APIServiceExportRequestReconciler) Reconcile(ctx context.Context, req m
// Run the reconciliation logic
if err := r.reconciler.reconcile(ctx, client, cache, apiServiceExportRequest); err != nil {
logger.Error(err, "Failed to reconcile APIServiceExportRequest")
if !reflect.DeepEqual(original.Status.Phase, apiServiceExportRequest.Status.Phase) {
if err := client.Status().Update(ctx, apiServiceExportRequest); err != nil {
logger.Error(err, "Failed to update APIServiceExportRequest status")
return ctrl.Result{}, fmt.Errorf("failed to update APIServiceExportRequest status: %w", err)
}
logger.Info("APIServiceExportRequest status updated", "namespace", apiServiceExportRequest.Namespace, "name", apiServiceExportRequest.Name)
}

return ctrl.Result{}, err
}

// Update status if it has changed
if !reflect.DeepEqual(original.Status, apiServiceExportRequest.Status) {
err := client.Status().Update(ctx, apiServiceExportRequest)
if err != nil {
if err := client.Status().Update(ctx, apiServiceExportRequest); err != nil {
logger.Error(err, "Failed to update APIServiceExportRequest status")
return ctrl.Result{}, fmt.Errorf("failed to update APIServiceExportRequest status: %w", err)
}
Expand All @@ -241,8 +247,8 @@ func (r *APIServiceExportRequestReconciler) SetupWithManager(mgr mcmanager.Manag
getServiceExportRequestMapper,
).
Watches(
&kubebindv1alpha2.APIResourceSchema{},
getAPIResourceSchemaMapper,
&kubebindv1alpha2.BoundSchema{},
getBoundSchemaMapper,
).
WithOptions(r.opts).
Named(controllerName).
Expand Down
Loading