From 572a893bbe0c531f3f4938ff1f73398cb4643b47 Mon Sep 17 00:00:00 2001 From: David Desmarais-Michaud Date: Sat, 6 Jun 2026 22:13:57 -0400 Subject: [PATCH] k8s/controller: expose informer cache --- internal/atc/reconciler_airway.go | 8 ++-- internal/atc/reconciler_flight.go | 9 +++-- internal/atc/reconciler_instance.go | 17 ++++---- pkg/k8s/ctrl/controller.go | 60 ++++++++++++++++++++++++++++- 4 files changed, 78 insertions(+), 16 deletions(-) diff --git a/internal/atc/reconciler_airway.go b/internal/atc/reconciler_airway.go index 22c9549b..1113dc75 100644 --- a/internal/atc/reconciler_airway.go +++ b/internal/atc/reconciler_airway.go @@ -24,6 +24,7 @@ import ( "github.com/yokecd/yoke/internal/k8s" "github.com/yokecd/yoke/internal/wasi/cache" "github.com/yokecd/yoke/internal/wasi/host" + "github.com/yokecd/yoke/pkg/apis/v1alpha1" "github.com/yokecd/yoke/pkg/flight" "github.com/yokecd/yoke/pkg/k8s/ctrl" "github.com/yokecd/yoke/pkg/openapi" @@ -33,10 +34,11 @@ func (atc atc) Reconcile(ctx context.Context, event ctrl.Event) (result ctrl.Res var ( client = (*k8s.Client)(ctrl.Client(ctx)) airwayIntf = client.AirwayIntf() + airwayCache = ctrl.CacheFromEvent[v1alpha1.Airway](ctx, event) webhookIntf = ctrl.Client(ctx).Clientset.AdmissionregistrationV1().ValidatingWebhookConfigurations() ) - airway, err := airwayIntf.Get(ctx, event.Name, metav1.GetOptions{}) + airway, err := airwayCache.Get(event.Name) if err != nil { if kerrors.IsNotFound(err) { ctrl.Logger(ctx).Info("airway not found") @@ -224,9 +226,9 @@ func (atc atc) Reconcile(ctx context.Context, event ctrl.Event) (result ctrl.Res } statusSchema, ok := version.Schema.OpenAPIV3Schema.Properties["status"] if !ok { - version.Schema.OpenAPIV3Schema.Properties["status"] = *(openapi.SchemaFor[struct { + version.Schema.OpenAPIV3Schema.Properties["status"] = *openapi.SchemaFor[struct { Conditions flight.Conditions `json:"conditions,omitempty"` - }]()) + }]() } else { if statusSchema.Type != "object" { return ctrl.Result{}, fmt.Errorf("invalid airway: status must be an object but got type: %q", statusSchema.Type) diff --git a/internal/atc/reconciler_flight.go b/internal/atc/reconciler_flight.go index 5144575d..47d1710b 100644 --- a/internal/atc/reconciler_flight.go +++ b/internal/atc/reconciler_flight.go @@ -47,16 +47,17 @@ func flightReconciler(modules *cache.ModuleCache, clusterScope bool) ctrl.Funcs type AltFlight v1alpha1.Flight var ( - client = (*k8s.Client)(ctrl.Client(ctx)) - commander = yoke.FromK8Client(client) - flightIntf = k8s.TypedInterface[AltFlight](client.Dynamic, gvr).Namespace(evt.Namespace) + client = (*k8s.Client)(ctrl.Client(ctx)) + commander = yoke.FromK8Client(client) + flightIntf = k8s.TypedInterface[AltFlight](client.Dynamic, gvr).Namespace(evt.Namespace) + flightCache = ctrl.CacheFromEvent[AltFlight](ctx, evt) ) if cleanup := cleanups[evt.String()]; cleanup != nil { cleanup() } - flight, err := flightIntf.Get(ctx, evt.Name, metav1.GetOptions{}) + flight, err := flightCache.Get(evt.Name) if err != nil { if kerrors.IsNotFound(err) { return ctrl.Result{}, nil diff --git a/internal/atc/reconciler_instance.go b/internal/atc/reconciler_instance.go index c4aed014..87df4bac 100644 --- a/internal/atc/reconciler_instance.go +++ b/internal/atc/reconciler_instance.go @@ -44,6 +44,7 @@ func (atc atc) InstanceReconciler(params InstanceReconcilerParams) ctrl.Funcs { reconciler := func(ctx context.Context, event ctrl.Event) (result ctrl.Result, err error) { client := (*k8s.Client)(ctrl.Client(ctx)) + resourceCache := ctrl.CacheFromEvent[unstructured.Unstructured](ctx, event) mapping, err := client.Mapper.RESTMapping(params.GK, params.Version) if err != nil { @@ -51,14 +52,7 @@ func (atc atc) InstanceReconciler(params InstanceReconcilerParams) ctrl.Funcs { return ctrl.Result{}, fmt.Errorf("failed to get rest mapping for gk: %w", err) } - resourceIntf := func() dynamic.ResourceInterface { - if mapping.Scope == meta.RESTScopeNamespace { - return client.Dynamic.Resource(mapping.Resource).Namespace(event.Namespace) - } - return client.Dynamic.Resource(mapping.Resource) - }() - - resource, err := resourceIntf.Get(ctx, event.Name, metav1.GetOptions{}) + resource, err := resourceCache.Get(event.Name) if err != nil { if kerrors.IsNotFound(err) { ctrl.Logger(ctx).Info("resource not found") @@ -67,6 +61,13 @@ func (atc atc) InstanceReconciler(params InstanceReconcilerParams) ctrl.Funcs { return ctrl.Result{}, fmt.Errorf("failed to get resource: %w", err) } + resourceIntf := func() dynamic.ResourceInterface { + if mapping.Scope == meta.RESTScopeNamespace { + return client.Dynamic.Resource(mapping.Resource).Namespace(event.Namespace) + } + return client.Dynamic.Resource(mapping.Resource) + }() + if resource.GetNamespace() == "" && mapping.Scope == meta.RESTScopeNamespace { resource.SetNamespace("default") if _, err := resourceIntf.Update(ctx, resource, metav1.UpdateOptions{FieldManager: fieldManager}); err != nil { diff --git a/pkg/k8s/ctrl/controller.go b/pkg/k8s/ctrl/controller.go index b4c418fd..5ec13115 100644 --- a/pkg/k8s/ctrl/controller.go +++ b/pkg/k8s/ctrl/controller.go @@ -16,11 +16,14 @@ import ( "github.com/davidmdm/x/xsync" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/client-go/dynamic/dynamicinformer" kcache "k8s.io/client-go/tools/cache" "github.com/yokecd/yoke/internal" + internalk8s "github.com/yokecd/yoke/internal/k8s" "github.com/yokecd/yoke/pkg/k8s" ) @@ -56,6 +59,7 @@ type HandleFunc func(context.Context, Event) (Result, error) type gkstate struct { handler HandleFunc + lister kcache.GenericLister shutdown func() } @@ -111,7 +115,9 @@ func (instance *Instance) register(entry Entry) error { factory := dynamicinformer.NewDynamicSharedInformerFactory(instance.Client.Dynamic, 0) - informer := factory.ForResource(mapping.Resource).Informer() + genericInformer := factory.ForResource(mapping.Resource) + informer := genericInformer.Informer() + lister := genericInformer.Lister() var resourceMap xsync.Set[Event] @@ -196,6 +202,7 @@ func (instance *Instance) register(entry Entry) error { teardown() } }), + lister: lister, }, ) @@ -387,3 +394,54 @@ func safe(handler HandleFunc) HandleFunc { return handler(ctx, event) } } + +type ResourceCache[T any] struct { + lister kcache.GenericNamespaceLister +} + +func (r ResourceCache[T]) Get(name string) (*T, error) { + obj, err := r.lister.Get(name) + if err != nil { + return nil, err + } + var result T + if err := runtime.DefaultUnstructuredConverter.FromUnstructured(obj.(*unstructured.Unstructured).Object, &result); err != nil { + return nil, err + } + return &result, nil +} + +func (r ResourceCache[T]) List(selector labels.Selector) ([]*T, error) { + objects, err := r.lister.List(selector) + if err != nil { + return nil, err + } + result := make([]*T, len(objects)) + for i, obj := range objects { + var value T + if err := runtime.DefaultUnstructuredConverter.FromUnstructured(obj.(*unstructured.Unstructured).Object, &value); err != nil { + return nil, err + } + result[i] = &value + } + return result, nil +} + +func Cache[T any, obj internalk8s.MetaObject[T]](ctx context.Context, gk schema.GroupKind, ns string) *ResourceCache[T] { + state, loaded := Inst(ctx).gks.Load(gk) + if !loaded { + return nil + } + return &ResourceCache[T]{ + lister: func() kcache.GenericNamespaceLister { + if ns != "" { + return state.lister.ByNamespace(ns) + } + return state.lister + }(), + } +} + +func CacheFromEvent[T any, obj internalk8s.MetaObject[T]](ctx context.Context, evt Event) *ResourceCache[T] { + return Cache[T, obj](ctx, evt.GroupKind, evt.Namespace) +}