88 lines
2.6 KiB
Go
88 lines
2.6 KiB
Go
package object
|
|
|
|
import (
|
|
"fmt"
|
|
|
|
meta "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/apimachinery/pkg/runtime"
|
|
"k8s.io/client-go/tools/cache"
|
|
)
|
|
|
|
// NewIndexerInformer is a copy of the cache.NewIndexerInformer function, but allows custom process function
|
|
func NewIndexerInformer(lw cache.ListerWatcher, objType runtime.Object, h cache.ResourceEventHandler, indexers cache.Indexers, builder ProcessorBuilder) (cache.Indexer, cache.Controller) {
|
|
clientState := cache.NewIndexer(cache.DeletionHandlingMetaNamespaceKeyFunc, indexers)
|
|
|
|
cfg := &cache.Config{
|
|
Queue: cache.NewDeltaFIFOWithOptions(cache.DeltaFIFOOptions{KeyFunction: cache.MetaNamespaceKeyFunc, KnownObjects: clientState}),
|
|
ListerWatcher: lw,
|
|
ObjectType: objType,
|
|
FullResyncPeriod: defaultResyncPeriod,
|
|
RetryOnError: false,
|
|
Process: builder(clientState, h),
|
|
}
|
|
return clientState, cache.New(cfg)
|
|
}
|
|
|
|
// RecordLatencyFunc is a function for recording api object delta latency
|
|
type RecordLatencyFunc func(meta.Object)
|
|
|
|
// DefaultProcessor is based on the Process function from cache.NewIndexerInformer except it does a conversion.
|
|
func DefaultProcessor(convert ToFunc, recordLatency *EndpointLatencyRecorder) ProcessorBuilder {
|
|
return func(clientState cache.Indexer, h cache.ResourceEventHandler) cache.ProcessFunc {
|
|
return func(obj interface{}, isInitialList bool) error {
|
|
for _, d := range obj.(cache.Deltas) {
|
|
if recordLatency != nil {
|
|
if o, ok := d.Object.(meta.Object); ok {
|
|
recordLatency.init(o)
|
|
}
|
|
}
|
|
switch d.Type {
|
|
case cache.Sync, cache.Added, cache.Updated:
|
|
obj, err := convert(d.Object.(meta.Object))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if old, exists, err := clientState.Get(obj); err == nil && exists {
|
|
if err := clientState.Update(obj); err != nil {
|
|
return err
|
|
}
|
|
h.OnUpdate(old, obj)
|
|
} else {
|
|
if err := clientState.Add(obj); err != nil {
|
|
return err
|
|
}
|
|
h.OnAdd(obj, isInitialList)
|
|
}
|
|
if recordLatency != nil {
|
|
recordLatency.record()
|
|
}
|
|
case cache.Deleted:
|
|
var obj interface{}
|
|
obj, ok := d.Object.(cache.DeletedFinalStateUnknown)
|
|
if !ok {
|
|
var err error
|
|
metaObj, ok := d.Object.(meta.Object)
|
|
if !ok {
|
|
return fmt.Errorf("unexpected object %v", d.Object)
|
|
}
|
|
obj, err = convert(metaObj)
|
|
if err != nil && err != errPodTerminating {
|
|
return err
|
|
}
|
|
}
|
|
|
|
if err := clientState.Delete(obj); err != nil {
|
|
return err
|
|
}
|
|
h.OnDelete(obj)
|
|
if !ok && recordLatency != nil {
|
|
recordLatency.record()
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
}
|
|
}
|
|
|
|
const defaultResyncPeriod = 0
|