diff --git a/tools/cache/reflector.go b/tools/cache/reflector.go index c5488f8a..62749ed7 100644 --- a/tools/cache/reflector.go +++ b/tools/cache/reflector.go @@ -29,7 +29,9 @@ import ( apierrs "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/util/clock" "k8s.io/apimachinery/pkg/util/naming" utilnet "k8s.io/apimachinery/pkg/util/net" @@ -41,13 +43,22 @@ import ( "k8s.io/utils/trace" ) +const defaultExpectedTypeName = "" + // Reflector watches a specified resource and causes all changes to be reflected in the given store. type Reflector struct { // name identifies this reflector. By default it will be a file:line if possible. name string + // The name of the type we expect to place in the store. The name + // will be the stringification of expectedGVK if provided, and the + // stringification of expectedType otherwise. It is for display + // only, and should not be used for parsing or comparison. + expectedTypeName string // The type of object we expect to place in the store. expectedType reflect.Type + // The GVK of the object we expect to place in the store if unstructured. + expectedGVK *schema.GroupVersionKind // The destination to sync up with the watch source store Store // listerWatcher is used to perform lists and watches. @@ -100,14 +111,35 @@ func NewNamedReflector(name string, lw ListerWatcher, expectedType interface{}, name: name, listerWatcher: lw, store: store, - expectedType: reflect.TypeOf(expectedType), period: time.Second, resyncPeriod: resyncPeriod, clock: &clock.RealClock{}, } + r.setExpectedType(expectedType) return r } +func (r *Reflector) setExpectedType(expectedType interface{}) { + r.expectedType = reflect.TypeOf(expectedType) + if r.expectedType == nil { + r.expectedTypeName = defaultExpectedTypeName + return + } + + r.expectedTypeName = r.expectedType.String() + + if obj, ok := expectedType.(*unstructured.Unstructured); ok { + // Use gvk to check that watch event objects are of the desired type. + gvk := obj.GroupVersionKind() + if gvk.Empty() { + klog.V(4).Infof("Reflector from %s configured with expectedType of *unstructured.Unstructured with empty GroupVersionKind.", r.name) + return + } + r.expectedGVK = &gvk + r.expectedTypeName = gvk.String() + } +} + // internalPackages are packages that ignored when creating a default reflector name. These packages are in the common // call chains to NewReflector, so they'd be low entropy names for reflectors var internalPackages = []string{"client-go/tools/cache/"} @@ -115,7 +147,7 @@ var internalPackages = []string{"client-go/tools/cache/"} // Run starts a watch and handles watch events. Will restart the watch if it is closed. // Run will exit when stopCh is closed. func (r *Reflector) Run(stopCh <-chan struct{}) { - klog.V(3).Infof("Starting reflector %v (%s) from %s", r.expectedType, r.resyncPeriod, r.name) + klog.V(3).Infof("Starting reflector %v (%s) from %s", r.expectedTypeName, r.resyncPeriod, r.name) wait.Until(func() { if err := r.ListAndWatch(stopCh); err != nil { utilruntime.HandleError(err) @@ -150,7 +182,7 @@ func (r *Reflector) resyncChan() (<-chan time.Time, func() bool) { // and then use the resource version to watch. // It returns error if ListAndWatch didn't even try to initialize watch. func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error { - klog.V(3).Infof("Listing and watching %v from %s", r.expectedType, r.name) + klog.V(3).Infof("Listing and watching %v from %s", r.expectedTypeName, r.name) var resourceVersion string // Explicitly set "0" as resource version - it's fine for the List() @@ -191,7 +223,7 @@ func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error { case <-listCh: } if err != nil { - return fmt.Errorf("%s: Failed to list %v: %v", r.name, r.expectedType, err) + return fmt.Errorf("%s: Failed to list %v: %v", r.name, r.expectedTypeName, err) } initTrace.Step("Objects listed") listMetaInterface, err := meta.ListAccessor(list) @@ -270,9 +302,9 @@ func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error { case io.EOF: // watch closed normally case io.ErrUnexpectedEOF: - klog.V(1).Infof("%s: Watch for %v closed with unexpected EOF: %v", r.name, r.expectedType, err) + klog.V(1).Infof("%s: Watch for %v closed with unexpected EOF: %v", r.name, r.expectedTypeName, err) default: - utilruntime.HandleError(fmt.Errorf("%s: Failed to watch %v: %v", r.name, r.expectedType, err)) + utilruntime.HandleError(fmt.Errorf("%s: Failed to watch %v: %v", r.name, r.expectedTypeName, err)) } // If this is "connection refused" error, it means that most likely apiserver is not responsive. // It doesn't make sense to re-list all objects because most likely we will be able to restart @@ -289,9 +321,9 @@ func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error { if err != errorStopRequested { switch { case apierrs.IsResourceExpired(err): - klog.V(4).Infof("%s: watch of %v ended with: %v", r.name, r.expectedType, err) + klog.V(4).Infof("%s: watch of %v ended with: %v", r.name, r.expectedTypeName, err) default: - klog.Warningf("%s: watch of %v ended with: %v", r.name, r.expectedType, err) + klog.Warningf("%s: watch of %v ended with: %v", r.name, r.expectedTypeName, err) } } return nil @@ -331,9 +363,17 @@ loop: if event.Type == watch.Error { return apierrs.FromObject(event.Object) } - if e, a := r.expectedType, reflect.TypeOf(event.Object); e != nil && e != a { - utilruntime.HandleError(fmt.Errorf("%s: expected type %v, but watch event object had type %v", r.name, e, a)) - continue + if r.expectedType != nil { + if e, a := r.expectedType, reflect.TypeOf(event.Object); e != a { + utilruntime.HandleError(fmt.Errorf("%s: expected type %v, but watch event object had type %v", r.name, e, a)) + continue + } + } + if r.expectedGVK != nil { + if e, a := *r.expectedGVK, event.Object.GetObjectKind().GroupVersionKind(); e != a { + utilruntime.HandleError(fmt.Errorf("%s: expected gvk %v, but watch event object had gvk %v", r.name, e, a)) + continue + } } meta, err := meta.Accessor(event.Object) if err != nil { @@ -375,7 +415,7 @@ loop: if watchDuration < 1*time.Second && eventCount == 0 { return fmt.Errorf("very short watch: %s: Unexpected watch close - watch lasted less than a second and no items received", r.name) } - klog.V(4).Infof("%s: Watch close - %v total %v items received", r.name, r.expectedType, eventCount) + klog.V(4).Infof("%s: Watch close - %v total %v items received", r.name, r.expectedTypeName, eventCount) return nil } diff --git a/tools/cache/reflector_test.go b/tools/cache/reflector_test.go index a62638c3..1f28dfce 100644 --- a/tools/cache/reflector_test.go +++ b/tools/cache/reflector_test.go @@ -20,13 +20,16 @@ import ( "errors" "fmt" "math/rand" + "reflect" "strconv" "testing" "time" "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/watch" ) @@ -430,3 +433,58 @@ func TestReflectorWatchListPageSize(t *testing.T) { t.Errorf("Expected 10 results, got %d", len(results)) } } + +func TestReflectorSetExpectedType(t *testing.T) { + obj := &unstructured.Unstructured{} + gvk := schema.GroupVersionKind{ + Group: "mygroup", + Version: "v1", + Kind: "MyKind", + } + obj.SetGroupVersionKind(gvk) + testCases := map[string]struct { + inputType interface{} + expectedTypeName string + expectedType reflect.Type + expectedGVK *schema.GroupVersionKind + }{ + "Nil type": { + expectedTypeName: defaultExpectedTypeName, + }, + "Normal type": { + inputType: &v1.Pod{}, + expectedTypeName: "*v1.Pod", + expectedType: reflect.TypeOf(&v1.Pod{}), + }, + "Unstructured type without GVK": { + inputType: &unstructured.Unstructured{}, + expectedTypeName: "*unstructured.Unstructured", + expectedType: reflect.TypeOf(&unstructured.Unstructured{}), + }, + "Unstructured type with GVK": { + inputType: obj, + expectedTypeName: gvk.String(), + expectedType: reflect.TypeOf(&unstructured.Unstructured{}), + expectedGVK: &gvk, + }, + } + for testName, tc := range testCases { + t.Run(testName, func(t *testing.T) { + r := &Reflector{} + r.setExpectedType(tc.inputType) + if tc.expectedType != r.expectedType { + t.Fatalf("Expected expectedType %v, got %v", tc.expectedType, r.expectedType) + } + if tc.expectedTypeName != r.expectedTypeName { + t.Fatalf("Expected expectedTypeName %v, got %v", tc.expectedTypeName, r.expectedTypeName) + } + gvkNotEqual := (tc.expectedGVK == nil) != (r.expectedGVK == nil) + if tc.expectedGVK != nil && r.expectedGVK != nil { + gvkNotEqual = *tc.expectedGVK != *r.expectedGVK + } + if gvkNotEqual { + t.Fatalf("Expected expectedGVK %v, got %v", tc.expectedGVK, r.expectedGVK) + } + }) + } +}