Merge pull request #73071 from krzysied/reflector_trace

Adding trace to reflector initialization
pull/564/head
Kubernetes Prow Robot 2019-01-22 05:15:55 -08:00 committed by GitHub
commit fd0df59f5b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 71 additions and 38 deletions

View File

@ -60,6 +60,7 @@
allowedImports:
- k8s.io/api
- k8s.io/apimachinery
- k8s.io/apiserver/pkg/util/trace
- k8s.io/client-go
- k8s.io/klog
@ -82,6 +83,7 @@
- "./vendor/k8s.io/client-go/tools/remotecommand"
allowedImports:
- k8s.io/apimachinery
- k8s.io/apiserver/pkg/util/trace
- k8s.io/client-go
- k8s.io/klog

View File

@ -610,6 +610,10 @@
"ImportPath": "k8s.io/apimachinery/third_party/forked/golang/reflect",
"Rev": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
},
{
"ImportPath": "k8s.io/apiserver/pkg/util/trace",
"Rev": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
},
{
"ImportPath": "k8s.io/klog",
"Rev": "8139d8cb77af419532b33dfa7dd09fbc5f1d344f"

View File

@ -79,6 +79,7 @@ go_library(
"//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/watch:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/util/trace:go_default_library",
"//staging/src/k8s.io/client-go/rest:go_default_library",
"//staging/src/k8s.io/client-go/tools/pager:go_default_library",
"//staging/src/k8s.io/client-go/util/buffer:go_default_library",

View File

@ -40,6 +40,7 @@ import (
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/apimachinery/pkg/watch"
utiltrace "k8s.io/apiserver/pkg/util/trace"
"k8s.io/klog"
)
@ -175,46 +176,56 @@ func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
options := metav1.ListOptions{ResourceVersion: "0"}
r.metrics.numberOfLists.Inc()
start := r.clock.Now()
var list runtime.Object
var err error
listCh := make(chan struct{}, 1)
panicCh := make(chan interface{}, 1)
go func() {
defer func() {
if r := recover(); r != nil {
panicCh <- r
}
if err := func() error {
trace := utiltrace.New("Reflector_" + r.name + "_ListAndWatch")
defer trace.LogIfLong(10 * time.Second)
var list runtime.Object
var err error
listCh := make(chan struct{}, 1)
panicCh := make(chan interface{}, 1)
go func() {
defer func() {
if r := recover(); r != nil {
panicCh <- r
}
}()
list, err = r.listerWatcher.List(options)
close(listCh)
}()
list, err = r.listerWatcher.List(options)
close(listCh)
}()
select {
case <-stopCh:
select {
case <-stopCh:
return nil
case r := <-panicCh:
panic(r)
case <-listCh:
}
if err != nil {
return fmt.Errorf("%s: Failed to list %v: %v", r.name, r.expectedType, err)
}
trace.Step("Objects listed")
r.metrics.listDuration.Observe(time.Since(start).Seconds())
listMetaInterface, err := meta.ListAccessor(list)
if err != nil {
return fmt.Errorf("%s: Unable to understand list result %#v: %v", r.name, list, err)
}
resourceVersion = listMetaInterface.GetResourceVersion()
trace.Step("Resource version extracted")
items, err := meta.ExtractList(list)
if err != nil {
return fmt.Errorf("%s: Unable to understand list result %#v (%v)", r.name, list, err)
}
trace.Step("Objects extracted")
r.metrics.numberOfItemsInList.Observe(float64(len(items)))
if err := r.syncWith(items, resourceVersion); err != nil {
return fmt.Errorf("%s: Unable to sync list result: %v", r.name, err)
}
trace.Step("SyncWith done")
r.setLastSyncResourceVersion(resourceVersion)
trace.Step("Resource version updated")
return nil
case r := <-panicCh:
panic(r)
case <-listCh:
}(); err != nil {
return err
}
if err != nil {
return fmt.Errorf("%s: Failed to list %v: %v", r.name, r.expectedType, err)
}
r.metrics.listDuration.Observe(time.Since(start).Seconds())
listMetaInterface, err := meta.ListAccessor(list)
if err != nil {
return fmt.Errorf("%s: Unable to understand list result %#v: %v", r.name, list, err)
}
resourceVersion = listMetaInterface.GetResourceVersion()
items, err := meta.ExtractList(list)
if err != nil {
return fmt.Errorf("%s: Unable to understand list result %#v (%v)", r.name, list, err)
}
r.metrics.numberOfItemsInList.Observe(float64(len(items)))
if err := r.syncWith(items, resourceVersion); err != nil {
return fmt.Errorf("%s: Unable to sync list result: %v", r.name, err)
}
r.setLastSyncResourceVersion(resourceVersion)
resyncerrc := make(chan error, 1)
cancelCh := make(chan struct{})
@ -285,7 +296,6 @@ func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
}
return nil
}
if err := r.watchHandler(w, &resourceVersion, resyncerrc, stopCh); err != nil {
if err != errorStopRequested {
klog.Warningf("%s: watch of %v ended with: %v", r.name, r.expectedType, err)

View File

@ -446,6 +446,10 @@
"ImportPath": "k8s.io/apimachinery/third_party/forked/golang/reflect",
"Rev": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
},
{
"ImportPath": "k8s.io/apiserver/pkg/util/trace",
"Rev": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
},
{
"ImportPath": "k8s.io/client-go/discovery",
"Rev": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

View File

@ -462,6 +462,10 @@
"ImportPath": "k8s.io/apimachinery/third_party/forked/golang/reflect",
"Rev": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
},
{
"ImportPath": "k8s.io/apiserver/pkg/util/trace",
"Rev": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
},
{
"ImportPath": "k8s.io/client-go/discovery",
"Rev": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

View File

@ -462,6 +462,10 @@
"ImportPath": "k8s.io/apimachinery/third_party/forked/golang/reflect",
"Rev": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
},
{
"ImportPath": "k8s.io/apiserver/pkg/util/trace",
"Rev": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
},
{
"ImportPath": "k8s.io/client-go/discovery",
"Rev": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

View File

@ -474,6 +474,10 @@
"ImportPath": "k8s.io/apimachinery/third_party/forked/golang/reflect",
"Rev": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
},
{
"ImportPath": "k8s.io/apiserver/pkg/util/trace",
"Rev": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
},
{
"ImportPath": "k8s.io/client-go/discovery",
"Rev": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"