212 lines
7.4 KiB
Go
212 lines
7.4 KiB
Go
/*
|
|
Copyright 2016 The Kubernetes Authors.
|
|
|
|
Licensed under the Apache License, Version 2.0 (the "License");
|
|
you may not use this file except in compliance with the License.
|
|
You may obtain a copy of the License at
|
|
|
|
http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
Unless required by applicable law or agreed to in writing, software
|
|
distributed under the License is distributed on an "AS IS" BASIS,
|
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
See the License for the specific language governing permissions and
|
|
limitations under the License.
|
|
*/
|
|
|
|
package storage
|
|
|
|
import (
|
|
"fmt"
|
|
"reflect"
|
|
"strconv"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"k8s.io/api/core/v1"
|
|
"k8s.io/apimachinery/pkg/api/meta"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/apimachinery/pkg/fields"
|
|
"k8s.io/apimachinery/pkg/labels"
|
|
"k8s.io/apimachinery/pkg/runtime"
|
|
"k8s.io/apimachinery/pkg/util/diff"
|
|
"k8s.io/apimachinery/pkg/util/wait"
|
|
"k8s.io/apimachinery/pkg/watch"
|
|
)
|
|
|
|
// verifies the cacheWatcher.process goroutine is properly cleaned up even if
|
|
// the writes to cacheWatcher.result channel is blocked.
|
|
func TestCacheWatcherCleanupNotBlockedByResult(t *testing.T) {
|
|
var lock sync.RWMutex
|
|
count := 0
|
|
filter := func(string, labels.Set, fields.Set, bool) bool { return true }
|
|
forget := func(bool) {
|
|
lock.Lock()
|
|
defer lock.Unlock()
|
|
count++
|
|
}
|
|
initEvents := []*watchCacheEvent{
|
|
{Object: &v1.Pod{}},
|
|
{Object: &v1.Pod{}},
|
|
}
|
|
// set the size of the buffer of w.result to 0, so that the writes to
|
|
// w.result is blocked.
|
|
w := newCacheWatcher(0, 0, initEvents, filter, forget, testVersioner{})
|
|
w.Stop()
|
|
if err := wait.PollImmediate(1*time.Second, 5*time.Second, func() (bool, error) {
|
|
lock.RLock()
|
|
defer lock.RUnlock()
|
|
return count == 2, nil
|
|
}); err != nil {
|
|
t.Fatalf("expected forget() to be called twice, because sendWatchCacheEvent should not be blocked by the result channel: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestCacheWatcherHandlesFiltering(t *testing.T) {
|
|
filter := func(_ string, _ labels.Set, field fields.Set, _ bool) bool {
|
|
return field["spec.nodeName"] == "host"
|
|
}
|
|
forget := func(bool) {}
|
|
|
|
testCases := []struct {
|
|
events []*watchCacheEvent
|
|
expected []watch.Event
|
|
}{
|
|
// properly handle starting with the filter, then being deleted, then re-added
|
|
{
|
|
events: []*watchCacheEvent{
|
|
{
|
|
Type: watch.Added,
|
|
Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "1"}},
|
|
ObjFields: fields.Set{"spec.nodeName": "host"},
|
|
ResourceVersion: 1,
|
|
},
|
|
{
|
|
Type: watch.Modified,
|
|
PrevObject: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "1"}},
|
|
PrevObjFields: fields.Set{"spec.nodeName": "host"},
|
|
Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "2"}},
|
|
ObjFields: fields.Set{"spec.nodeName": ""},
|
|
ResourceVersion: 2,
|
|
},
|
|
{
|
|
Type: watch.Modified,
|
|
PrevObject: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "2"}},
|
|
PrevObjFields: fields.Set{"spec.nodeName": ""},
|
|
Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "3"}},
|
|
ObjFields: fields.Set{"spec.nodeName": "host"},
|
|
ResourceVersion: 3,
|
|
},
|
|
},
|
|
expected: []watch.Event{
|
|
{Type: watch.Added, Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "1"}}},
|
|
{Type: watch.Deleted, Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "2"}}},
|
|
{Type: watch.Added, Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "3"}}},
|
|
},
|
|
},
|
|
// properly handle ignoring changes prior to the filter, then getting added, then deleted
|
|
{
|
|
events: []*watchCacheEvent{
|
|
{
|
|
Type: watch.Added,
|
|
Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "1"}},
|
|
ObjFields: fields.Set{"spec.nodeName": ""},
|
|
ResourceVersion: 1,
|
|
},
|
|
{
|
|
Type: watch.Modified,
|
|
PrevObject: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "1"}},
|
|
PrevObjFields: fields.Set{"spec.nodeName": ""},
|
|
Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "2"}},
|
|
ObjFields: fields.Set{"spec.nodeName": ""},
|
|
ResourceVersion: 2,
|
|
},
|
|
{
|
|
Type: watch.Modified,
|
|
PrevObject: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "2"}},
|
|
PrevObjFields: fields.Set{"spec.nodeName": ""},
|
|
Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "3"}},
|
|
ObjFields: fields.Set{"spec.nodeName": "host"},
|
|
ResourceVersion: 3,
|
|
},
|
|
{
|
|
Type: watch.Modified,
|
|
PrevObject: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "3"}},
|
|
PrevObjFields: fields.Set{"spec.nodeName": "host"},
|
|
Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "4"}},
|
|
ObjFields: fields.Set{"spec.nodeName": "host"},
|
|
ResourceVersion: 4,
|
|
},
|
|
{
|
|
Type: watch.Modified,
|
|
PrevObject: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "4"}},
|
|
PrevObjFields: fields.Set{"spec.nodeName": "host"},
|
|
Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "5"}},
|
|
ObjFields: fields.Set{"spec.nodeName": ""},
|
|
ResourceVersion: 5,
|
|
},
|
|
{
|
|
Type: watch.Modified,
|
|
PrevObject: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "5"}},
|
|
PrevObjFields: fields.Set{"spec.nodeName": ""},
|
|
Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "6"}},
|
|
ObjFields: fields.Set{"spec.nodeName": ""},
|
|
ResourceVersion: 6,
|
|
},
|
|
},
|
|
expected: []watch.Event{
|
|
{Type: watch.Added, Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "3"}}},
|
|
{Type: watch.Modified, Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "4"}}},
|
|
{Type: watch.Deleted, Object: &v1.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: "5"}}},
|
|
},
|
|
},
|
|
}
|
|
|
|
TestCase:
|
|
for i, testCase := range testCases {
|
|
// set the size of the buffer of w.result to 0, so that the writes to
|
|
// w.result is blocked.
|
|
for j := range testCase.events {
|
|
testCase.events[j].ResourceVersion = uint64(j) + 1
|
|
}
|
|
w := newCacheWatcher(0, 0, testCase.events, filter, forget, testVersioner{})
|
|
ch := w.ResultChan()
|
|
for j, event := range testCase.expected {
|
|
e := <-ch
|
|
if !reflect.DeepEqual(event, e) {
|
|
t.Errorf("%d: unexpected event %d: %s", i, j, diff.ObjectReflectDiff(event, e))
|
|
break TestCase
|
|
}
|
|
}
|
|
select {
|
|
case obj, ok := <-ch:
|
|
t.Errorf("%d: unexpected excess event: %#v %t", i, obj, ok)
|
|
break TestCase
|
|
default:
|
|
}
|
|
w.Stop()
|
|
}
|
|
}
|
|
|
|
type testVersioner struct{}
|
|
|
|
func (testVersioner) UpdateObject(obj runtime.Object, resourceVersion uint64) error {
|
|
return meta.NewAccessor().SetResourceVersion(obj, strconv.FormatUint(resourceVersion, 10))
|
|
}
|
|
func (testVersioner) UpdateList(obj runtime.Object, resourceVersion uint64, continueValue string) error {
|
|
return fmt.Errorf("unimplemented")
|
|
}
|
|
func (testVersioner) PrepareObjectForStorage(obj runtime.Object) error {
|
|
return fmt.Errorf("unimplemented")
|
|
}
|
|
func (testVersioner) ObjectResourceVersion(obj runtime.Object) (uint64, error) {
|
|
return 0, fmt.Errorf("unimplemented")
|
|
}
|
|
func (testVersioner) ParseWatchResourceVersion(resourceVersion string) (uint64, error) {
|
|
return strconv.ParseUint(resourceVersion, 10, 64)
|
|
}
|
|
func (testVersioner) ParseListResourceVersion(resourceVersion string) (uint64, error) {
|
|
return strconv.ParseUint(resourceVersion, 10, 64)
|
|
}
|