Alternative implementation that forks bits from kubernetes/client-go

Author: fabianvf

pkg/kcp/controller.go

@@ -0,0 +1,153 @@
+package kcp
+
+import (
+	"reflect"
+	"sync"
+	"time"
+	"unsafe"
+
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/utils/clock"
+)
+
+// `*controller` implements cache.Controller
+type controller struct {
+	config         cache.Config
+	reflector      *cache.Reflector
+	reflectorMutex sync.RWMutex
+	clock          clock.Clock
+}
+
+// New makes a new cache.Controller from the given cache.Config.
+func New(c *cache.Config) cache.Controller {
+	ctlr := &controller{
+		config: *c,
+		clock:  &clock.RealClock{},
+	}
+	return ctlr
+}
+
+// Run begins processing items, and will continue until a value is sent down stopCh or it is closed.
+// It's an error to call Run more than once.
+// Run blocks; call via go.
+func (c *controller) Run(stopCh <-chan struct{}) {
+	defer utilruntime.HandleCrash()
+	go func() {
+		<-stopCh
+		c.config.Queue.Close()
+	}()
+	r := cache.NewReflector(
+		c.config.ListerWatcher,
+		c.config.ObjectType,
+		c.config.Queue,
+		c.config.FullResyncPeriod,
+	)
+	r.ShouldResync = c.config.ShouldResync
+	r.WatchListPageSize = c.config.WatchListPageSize
+
+	reflector := reflect.Indirect(reflect.ValueOf(r))
+	clockField := reflector.FieldByName("clock")
+	ptrToClock := unsafe.Pointer(clockField.UnsafeAddr())
+	realPtrToClock := (*clock.Clock)(ptrToClock)
+	*realPtrToClock = c.clock
+	if c.config.WatchErrorHandler != nil {
+		handlerField := reflector.FieldByName("watchErrorHandler")
+		ptrToHandler := unsafe.Pointer(handlerField.UnsafeAddr())
+		realPtrToHandler := (*cache.WatchErrorHandler)(ptrToHandler)
+		*realPtrToHandler = c.config.WatchErrorHandler
+	}
+
+	c.reflectorMutex.Lock()
+	c.reflector = r
+	c.reflectorMutex.Unlock()
+
+	var wg wait.Group
+
+	wg.StartWithChannel(stopCh, r.Run)
+
+	wait.Until(c.processLoop, time.Second, stopCh)
+	wg.Wait()
+}
+
+// Returns true once this controller has completed an initial resource listing
+func (c *controller) HasSynced() bool {
+	return c.config.Queue.HasSynced()
+}
+
+func (c *controller) LastSyncResourceVersion() string {
+	c.reflectorMutex.RLock()
+	defer c.reflectorMutex.RUnlock()
+	if c.reflector == nil {
+		return ""
+	}
+	return c.reflector.LastSyncResourceVersion()
+}
+
+// processLoop drains the work queue.
+// TODO: Consider doing the processing in parallel. This will require a little thought
+// to make sure that we don't end up processing the same object multiple times
+// concurrently.
+//
+// TODO: Plumb through the stopCh here (and down to the queue) so that this can
+// actually exit when the controller is stopped. Or just give up on this stuff
+// ever being stoppable. Converting this whole package to use Context would
+// also be helpful.
+func (c *controller) processLoop() {
+	for {
+		obj, err := c.config.Queue.Pop(cache.PopProcessFunc(c.config.Process))
+		if err != nil {
+			if err == cache.ErrFIFOClosed {
+				return
+			}
+			if c.config.RetryOnError {
+				// This is the safe way to re-enqueue.
+				c.config.Queue.AddIfNotPresent(obj)
+			}
+		}
+	}
+}
+
+// Multiplexes updates in the form of a list of cache.Deltas into a cache.Store, and informs
+// a given handler of events OnUpdate, OnAdd, OnDelete
+func processDeltas(
+	// Object which receives event notifications from the given deltas
+	handler cache.ResourceEventHandler,
+	clientState cache.Store,
+	transformer cache.TransformFunc,
+	deltas cache.Deltas,
+) error {
+	// from oldest to newest
+	for _, d := range deltas {
+		obj := d.Object
+		if transformer != nil {
+			var err error
+			obj, err = transformer(obj)
+			if err != nil {
+				return err
+			}
+		}
+
+		switch d.Type {
+		case cache.Sync, cache.Replaced, cache.Added, cache.Updated:
+			if old, exists, err := clientState.Get(obj); err == nil && exists {
+				if err := clientState.Update(obj); err != nil {
+					return err
+				}
+				handler.OnUpdate(old, obj)
+			} else {
+				if err := clientState.Add(obj); err != nil {
+					return err
+				}
+				handler.OnAdd(obj)
+			}
+		case cache.Deleted:
+			if err := clientState.Delete(obj); err != nil {
+				return err
+			}
+			handler.OnDelete(obj)
+		}
+	}
+	return nil
+}