aboutsummaryrefslogtreecommitdiff
path: root/plugin/kubernetes/object/informer.go
diff options
context:
space:
mode:
Diffstat (limited to 'plugin/kubernetes/object/informer.go')
-rw-r--r--plugin/kubernetes/object/informer.go21
1 files changed, 16 insertions, 5 deletions
diff --git a/plugin/kubernetes/object/informer.go b/plugin/kubernetes/object/informer.go
index afd134e56..e44cd218a 100644
--- a/plugin/kubernetes/object/informer.go
+++ b/plugin/kubernetes/object/informer.go
@@ -1,6 +1,8 @@
package object
import (
+ "fmt"
+
meta "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/client-go/tools/cache"
@@ -25,13 +27,18 @@ func NewIndexerInformer(lw cache.ListerWatcher, objType runtime.Object, h cache.
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 RecordLatencyFunc) ProcessorBuilder {
+func DefaultProcessor(convert ToFunc, recordLatency *EndpointLatencyRecorder) ProcessorBuilder {
return func(clientState cache.Indexer, h cache.ResourceEventHandler) cache.ProcessFunc {
return func(obj interface{}) 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)
+ obj, err := convert(d.Object.(meta.Object))
if err != nil {
return err
}
@@ -47,14 +54,18 @@ func DefaultProcessor(convert ToFunc, recordLatency RecordLatencyFunc) Processor
h.OnAdd(obj)
}
if recordLatency != nil {
- recordLatency(d.Object.(meta.Object))
+ recordLatency.record()
}
case cache.Deleted:
var obj interface{}
obj, ok := d.Object.(cache.DeletedFinalStateUnknown)
if !ok {
var err error
- obj, err = convert(d.Object)
+ 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
}
@@ -65,7 +76,7 @@ func DefaultProcessor(convert ToFunc, recordLatency RecordLatencyFunc) Processor
}
h.OnDelete(obj)
if !ok && recordLatency != nil {
- recordLatency(d.Object.(meta.Object))
+ recordLatency.record()
}
}
}