coredns/plugin/kubernetes/object/informer.go
janluk d7cdb992b4 Measure and expose DNS programming latency from Kubernetes plugin. (#3171)
For now metric is measure only for headless services. Informer has been slighlty
refactored, so the code can measure latency without storing extra fields on
Endpoint struct.

Signed-off-by: Janek Łukaszewicz <janluk@google.com>

Suggestions from code review

Co-Authored-By: Chris O'Haver <cohaver@infoblox.com>
2019-10-04 16:48:43 +01:00

56 lines
1.7 KiB
Go

package object
import (
"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.NewDeltaFIFO(cache.MetaNamespaceKeyFunc, clientState),
ListerWatcher: lw,
ObjectType: objType,
FullResyncPeriod: defaultResyncPeriod,
RetryOnError: false,
Process: builder(clientState, h),
}
return clientState, cache.New(cfg)
}
// DefaultProcessor is a copy of Process function from cache.NewIndexerInformer except it does a conversion.
func DefaultProcessor(convert ToFunc) ProcessorBuilder {
return func(clientState cache.Indexer, h cache.ResourceEventHandler) cache.ProcessFunc {
return func(obj interface{}) error {
for _, d := range obj.(cache.Deltas) {
obj := convert(d.Object)
switch d.Type {
case cache.Sync, cache.Added, cache.Updated:
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)
}
case cache.Deleted:
if err := clientState.Delete(obj); err != nil {
return err
}
h.OnDelete(obj)
}
}
return nil
}
}
}
const defaultResyncPeriod = 0