Add hacky kcp informer func

Author: fabianvf

pkg/client/interfaces.go

@@ -18,10 +18,12 @@ package client
 
 import (
 	"context"
+	"time"
 
 	"github.com/kcp-dev/logicalcluster"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/client-go/tools/cache"
 
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -48,6 +50,8 @@ type Patch interface {
 	Data(obj Object) ([]byte, error)
 }
 
+type NewInformerFunc func(cache.ListerWatcher, runtime.Object, time.Duration, cache.Indexers) cache.SharedIndexInformer
+
 // TODO(directxman12): is there a sane way to deal with get/delete options?
 
 // Reader knows how to read and list Kubernetes objects.

pkg/kcp/informers.go

@@ -0,0 +1,108 @@
+package kcp
+
+import (
+	"reflect"
+	"time"
+	"unsafe"
+
+	k8sruntime "k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/tools/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+type clusterAwareSharedIndexInformer struct {
+	delegate cache.SharedIndexInformer
+	keyFunc  cache.KeyFunc
+}
+
+func NewClusterAwareSharedIndexInformerFunc(keyFunc cache.KeyFunc) client.NewInformerFunc {
+	//TODO(kcp) could hardcode kcp indexers here or leave them plumbed through so they are passed when this function is called
+	return func(lw cache.ListerWatcher, exampleObject k8sruntime.Object, defaultEventHandlerResyncPeriod time.Duration, indexers cache.Indexers) cache.SharedIndexInformer {
+		return &clusterAwareSharedIndexInformer{
+			delegate: cache.NewSharedIndexInformer(lw, exampleObject, defaultEventHandlerResyncPeriod, indexers),
+			keyFunc:  keyFunc,
+		}
+	}
+}
+
+func (i *clusterAwareSharedIndexInformer) AddEventHandler(handler cache.ResourceEventHandler) {
+	i.delegate.AddEventHandler(handler)
+}
+
+func (i *clusterAwareSharedIndexInformer) AddEventHandlerWithResyncPeriod(handler cache.ResourceEventHandler, resyncPeriod time.Duration) {
+	i.delegate.AddEventHandlerWithResyncPeriod(handler, resyncPeriod)
+}
+
+func (i *clusterAwareSharedIndexInformer) GetStore() cache.Store {
+	return i.delegate.GetStore()
+}
+
+func (i *clusterAwareSharedIndexInformer) GetController() cache.Controller {
+	return i.delegate.GetController()
+}
+
+func (i *clusterAwareSharedIndexInformer) Run(stopCh <-chan struct{}) {
+	go i.delegate.Run(stopCh)
+	// TODO(kcp) this is fragile
+	time.Sleep(10000)
+	// need to update keyfunc in:
+	// [x] i.delegate.indexer.keyFunc
+	// [x] i.delegate.controller.config.Queue.keyFunc
+	// [x] i.delegate.controller.reflector.store.keyFunc
+
+	indexer := i.delegate.GetIndexer()
+	i.setKeyFunc(indexer)
+
+	pointerVal := reflect.ValueOf(i.delegate)
+	val := reflect.Indirect(pointerVal)
+	controllerField := val.FieldByName("controller")
+	ptrToController := unsafe.Pointer(controllerField.UnsafeAddr())
+	controller := reflect.NewAt(controllerField.Type(), ptrToController).Elem().Interface()
+
+	configval := reflect.Indirect(reflect.ValueOf(controller)).FieldByName("config")
+	ptrToConfig := unsafe.Pointer(configval.UnsafeAddr())
+	config := (*cache.Config)(ptrToConfig)
+	deltaFifo := config.Queue.(*cache.DeltaFIFO)
+	i.setKeyFunc(deltaFifo)
+
+	reflectorval := reflect.Indirect(reflect.ValueOf(controller)).FieldByName("reflector")
+	ptrToReflector := unsafe.Pointer(reflectorval.UnsafeAddr())
+	reflector := (*cache.Reflector)(ptrToReflector)
+
+	storeval := reflect.Indirect(reflect.ValueOf(reflector)).FieldByName("store")
+	ptrToStore := unsafe.Pointer(storeval.UnsafeAddr())
+	deltaFifo = (*cache.DeltaFIFO)(ptrToStore)
+	i.setKeyFunc(deltaFifo)
+
+	// deltaFifo := config.Queue.(*cache.DeltaFIFO)
+	// i.setKeyFunc(deltaFifo)
+}
+
+func (i *clusterAwareSharedIndexInformer) setKeyFunc(x interface{}) {
+	pointerVal := reflect.ValueOf(x)
+	val := reflect.Indirect(pointerVal)
+	member := val.FieldByName("keyFunc")
+	ptrToY := unsafe.Pointer(member.UnsafeAddr())
+	realPtrToY := (*cache.KeyFunc)(ptrToY)
+	*realPtrToY = i.keyFunc
+}
+
+func (i *clusterAwareSharedIndexInformer) HasSynced() bool {
+	return i.delegate.HasSynced()
+}
+
+func (i *clusterAwareSharedIndexInformer) LastSyncResourceVersion() string {
+	return i.delegate.LastSyncResourceVersion()
+}
+
+func (i *clusterAwareSharedIndexInformer) SetWatchErrorHandler(handler cache.WatchErrorHandler) error {
+	return i.delegate.SetWatchErrorHandler(handler)
+}
+
+func (i *clusterAwareSharedIndexInformer) AddIndexers(indexers cache.Indexers) error {
+	return i.delegate.AddIndexers(indexers)
+}
+
+func (i *clusterAwareSharedIndexInformer) GetIndexer() cache.Indexer {
+	return i.delegate.GetIndexer()
+}