444 lines
15 KiB
Go
444 lines
15 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 etcd3
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"go.etcd.io/etcd/api/v3/mvccpb"
|
|
clientv3 "go.etcd.io/etcd/client/v3"
|
|
|
|
"k8s.io/apimachinery/pkg/api/apitesting"
|
|
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/runtime/schema"
|
|
"k8s.io/apimachinery/pkg/util/wait"
|
|
"k8s.io/apimachinery/pkg/watch"
|
|
"k8s.io/apiserver/pkg/apis/example"
|
|
examplev1 "k8s.io/apiserver/pkg/apis/example/v1"
|
|
"k8s.io/apiserver/pkg/storage"
|
|
"k8s.io/apiserver/pkg/storage/etcd3/testserver"
|
|
utilflowcontrol "k8s.io/apiserver/pkg/util/flowcontrol"
|
|
)
|
|
|
|
func TestWatch(t *testing.T) {
|
|
testWatch(t, false)
|
|
testWatch(t, true)
|
|
}
|
|
|
|
// It tests that
|
|
// - first occurrence of objects should notify Add event
|
|
// - update should trigger Modified event
|
|
// - update that gets filtered should trigger Deleted event
|
|
func testWatch(t *testing.T, recursive bool) {
|
|
ctx, store, _ := testSetup(t)
|
|
podFoo := &example.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo"}}
|
|
podBar := &example.Pod{ObjectMeta: metav1.ObjectMeta{Name: "bar"}}
|
|
|
|
tests := []struct {
|
|
name string
|
|
key string
|
|
pred storage.SelectionPredicate
|
|
watchTests []*testWatchStruct
|
|
}{{
|
|
name: "create a key",
|
|
key: "/somekey-1",
|
|
watchTests: []*testWatchStruct{{podFoo, true, watch.Added}},
|
|
pred: storage.Everything,
|
|
}, {
|
|
name: "key updated to match predicate",
|
|
key: "/somekey-3",
|
|
watchTests: []*testWatchStruct{{podFoo, false, ""}, {podBar, true, watch.Added}},
|
|
pred: storage.SelectionPredicate{
|
|
Label: labels.Everything(),
|
|
Field: fields.ParseSelectorOrDie("metadata.name=bar"),
|
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
|
pod := obj.(*example.Pod)
|
|
return nil, fields.Set{"metadata.name": pod.Name}, nil
|
|
},
|
|
},
|
|
}, {
|
|
name: "update",
|
|
key: "/somekey-4",
|
|
watchTests: []*testWatchStruct{{podFoo, true, watch.Added}, {podBar, true, watch.Modified}},
|
|
pred: storage.Everything,
|
|
}, {
|
|
name: "delete because of being filtered",
|
|
key: "/somekey-5",
|
|
watchTests: []*testWatchStruct{{podFoo, true, watch.Added}, {podBar, true, watch.Deleted}},
|
|
pred: storage.SelectionPredicate{
|
|
Label: labels.Everything(),
|
|
Field: fields.ParseSelectorOrDie("metadata.name!=bar"),
|
|
GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) {
|
|
pod := obj.(*example.Pod)
|
|
return nil, fields.Set{"metadata.name": pod.Name}, nil
|
|
},
|
|
},
|
|
}}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
w, err := store.Watch(ctx, tt.key, storage.ListOptions{ResourceVersion: "0", Predicate: tt.pred, Recursive: recursive})
|
|
if err != nil {
|
|
t.Fatalf("Watch failed: %v", err)
|
|
}
|
|
var prevObj *example.Pod
|
|
for _, watchTest := range tt.watchTests {
|
|
out := &example.Pod{}
|
|
key := tt.key
|
|
if recursive {
|
|
key = key + "/item"
|
|
}
|
|
err := store.GuaranteedUpdate(ctx, key, out, true, nil, storage.SimpleUpdate(
|
|
func(runtime.Object) (runtime.Object, error) {
|
|
return watchTest.obj, nil
|
|
}), nil)
|
|
if err != nil {
|
|
t.Fatalf("GuaranteedUpdate failed: %v", err)
|
|
}
|
|
if watchTest.expectEvent {
|
|
expectObj := out
|
|
if watchTest.watchType == watch.Deleted {
|
|
expectObj = prevObj
|
|
expectObj.ResourceVersion = out.ResourceVersion
|
|
}
|
|
testCheckResult(t, watchTest.watchType, w, expectObj)
|
|
}
|
|
prevObj = out
|
|
}
|
|
w.Stop()
|
|
testCheckStop(t, w)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestDeleteTriggerWatch(t *testing.T) {
|
|
ctx, store, _ := testSetup(t)
|
|
key, storedObj := testPropogateStore(ctx, t, store, &example.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo"}})
|
|
w, err := store.Watch(ctx, key, storage.ListOptions{ResourceVersion: storedObj.ResourceVersion, Predicate: storage.Everything})
|
|
if err != nil {
|
|
t.Fatalf("Watch failed: %v", err)
|
|
}
|
|
if err := store.Delete(ctx, key, &example.Pod{}, nil, storage.ValidateAllObjectFunc, nil); err != nil {
|
|
t.Fatalf("Delete failed: %v", err)
|
|
}
|
|
testCheckEventType(t, watch.Deleted, w)
|
|
}
|
|
|
|
// TestWatchFromZero tests that
|
|
// - watch from 0 should sync up and grab the object added before
|
|
// - watch from 0 is able to return events for objects whose previous version has been compacted
|
|
func TestWatchFromZero(t *testing.T) {
|
|
ctx, store, client := testSetup(t)
|
|
key, storedObj := testPropogateStore(ctx, t, store, &example.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: "ns"}})
|
|
|
|
w, err := store.Watch(ctx, key, storage.ListOptions{ResourceVersion: "0", Predicate: storage.Everything})
|
|
if err != nil {
|
|
t.Fatalf("Watch failed: %v", err)
|
|
}
|
|
testCheckResult(t, watch.Added, w, storedObj)
|
|
w.Stop()
|
|
|
|
// Update
|
|
out := &example.Pod{}
|
|
err = store.GuaranteedUpdate(ctx, key, out, true, nil, storage.SimpleUpdate(
|
|
func(runtime.Object) (runtime.Object, error) {
|
|
return &example.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: "ns", Annotations: map[string]string{"a": "1"}}}, nil
|
|
}), nil)
|
|
if err != nil {
|
|
t.Fatalf("GuaranteedUpdate failed: %v", err)
|
|
}
|
|
|
|
// Make sure when we watch from 0 we receive an ADDED event
|
|
w, err = store.Watch(ctx, key, storage.ListOptions{ResourceVersion: "0", Predicate: storage.Everything})
|
|
if err != nil {
|
|
t.Fatalf("Watch failed: %v", err)
|
|
}
|
|
testCheckResult(t, watch.Added, w, out)
|
|
w.Stop()
|
|
|
|
// Update again
|
|
out = &example.Pod{}
|
|
err = store.GuaranteedUpdate(ctx, key, out, true, nil, storage.SimpleUpdate(
|
|
func(runtime.Object) (runtime.Object, error) {
|
|
return &example.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: "ns"}}, nil
|
|
}), nil)
|
|
if err != nil {
|
|
t.Fatalf("GuaranteedUpdate failed: %v", err)
|
|
}
|
|
|
|
// Compact previous versions
|
|
revToCompact, err := store.versioner.ParseResourceVersion(out.ResourceVersion)
|
|
if err != nil {
|
|
t.Fatalf("Error converting %q to an int: %v", storedObj.ResourceVersion, err)
|
|
}
|
|
_, err = client.Compact(ctx, int64(revToCompact), clientv3.WithCompactPhysical())
|
|
if err != nil {
|
|
t.Fatalf("Error compacting: %v", err)
|
|
}
|
|
|
|
// Make sure we can still watch from 0 and receive an ADDED event
|
|
w, err = store.Watch(ctx, key, storage.ListOptions{ResourceVersion: "0", Predicate: storage.Everything})
|
|
if err != nil {
|
|
t.Fatalf("Watch failed: %v", err)
|
|
}
|
|
testCheckResult(t, watch.Added, w, out)
|
|
}
|
|
|
|
// TestWatchFromNoneZero tests that
|
|
// - watch from non-0 should just watch changes after given version
|
|
func TestWatchFromNoneZero(t *testing.T) {
|
|
ctx, store, _ := testSetup(t)
|
|
key, storedObj := testPropogateStore(ctx, t, store, &example.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo"}})
|
|
|
|
w, err := store.Watch(ctx, key, storage.ListOptions{ResourceVersion: storedObj.ResourceVersion, Predicate: storage.Everything})
|
|
if err != nil {
|
|
t.Fatalf("Watch failed: %v", err)
|
|
}
|
|
out := &example.Pod{}
|
|
store.GuaranteedUpdate(ctx, key, out, true, nil, storage.SimpleUpdate(
|
|
func(runtime.Object) (runtime.Object, error) {
|
|
return &example.Pod{ObjectMeta: metav1.ObjectMeta{Name: "bar"}}, err
|
|
}), nil)
|
|
testCheckResult(t, watch.Modified, w, out)
|
|
}
|
|
|
|
func TestWatchError(t *testing.T) {
|
|
// this codec fails on decodes, which will bubble up so we can verify the behavior
|
|
invalidCodec := &testCodec{apitesting.TestCodec(codecs, examplev1.SchemeGroupVersion)}
|
|
client := testserver.RunEtcd(t, nil)
|
|
invalidStore := newStore(client, invalidCodec, newPod, "", schema.GroupResource{Resource: "pods"}, &prefixTransformer{prefix: []byte("test!")}, true, newTestLeaseManagerConfig())
|
|
ctx := context.Background()
|
|
w, err := invalidStore.Watch(ctx, "/abc", storage.ListOptions{ResourceVersion: "0", Predicate: storage.Everything})
|
|
if err != nil {
|
|
t.Fatalf("Watch failed: %v", err)
|
|
}
|
|
codec := apitesting.TestCodec(codecs, examplev1.SchemeGroupVersion)
|
|
validStore := newStore(client, codec, newPod, "", schema.GroupResource{Resource: "pods"}, &prefixTransformer{prefix: []byte("test!")}, true, newTestLeaseManagerConfig())
|
|
if err := validStore.GuaranteedUpdate(ctx, "/abc", &example.Pod{}, true, nil, storage.SimpleUpdate(
|
|
func(runtime.Object) (runtime.Object, error) {
|
|
return &example.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo"}}, nil
|
|
}), nil); err != nil {
|
|
t.Fatalf("GuaranteedUpdate failed: %v", err)
|
|
}
|
|
testCheckEventType(t, watch.Error, w)
|
|
}
|
|
|
|
func TestWatchContextCancel(t *testing.T) {
|
|
ctx, store, _ := testSetup(t)
|
|
canceledCtx, cancel := context.WithCancel(ctx)
|
|
cancel()
|
|
// When we watch with a canceled context, we should detect that it's context canceled.
|
|
// We won't take it as error and also close the watcher.
|
|
w, err := store.watcher.Watch(canceledCtx, "/abc", 0, false, false, storage.Everything)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
select {
|
|
case _, ok := <-w.ResultChan():
|
|
if ok {
|
|
t.Error("ResultChan() should be closed")
|
|
}
|
|
case <-time.After(wait.ForeverTestTimeout):
|
|
t.Errorf("timeout after %v", wait.ForeverTestTimeout)
|
|
}
|
|
}
|
|
|
|
func TestWatchErrResultNotBlockAfterCancel(t *testing.T) {
|
|
origCtx, store, _ := testSetup(t)
|
|
ctx, cancel := context.WithCancel(origCtx)
|
|
w := store.watcher.createWatchChan(ctx, "/abc", 0, false, false, storage.Everything)
|
|
// make resutlChan and errChan blocking to ensure ordering.
|
|
w.resultChan = make(chan watch.Event)
|
|
w.errChan = make(chan error)
|
|
// The event flow goes like:
|
|
// - first we send an error, it should block on resultChan.
|
|
// - Then we cancel ctx. The blocking on resultChan should be freed up
|
|
// and run() goroutine should return.
|
|
var wg sync.WaitGroup
|
|
wg.Add(1)
|
|
go func() {
|
|
w.run()
|
|
wg.Done()
|
|
}()
|
|
w.errChan <- fmt.Errorf("some error")
|
|
cancel()
|
|
wg.Wait()
|
|
}
|
|
|
|
func TestWatchDeleteEventObjectHaveLatestRV(t *testing.T) {
|
|
ctx, store, client := testSetup(t)
|
|
key, storedObj := testPropogateStore(ctx, t, store, &example.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo"}})
|
|
|
|
w, err := store.Watch(ctx, key, storage.ListOptions{ResourceVersion: storedObj.ResourceVersion, Predicate: storage.Everything})
|
|
if err != nil {
|
|
t.Fatalf("Watch failed: %v", err)
|
|
}
|
|
rv, err := APIObjectVersioner{}.ObjectResourceVersion(storedObj)
|
|
if err != nil {
|
|
t.Fatalf("failed to parse resourceVersion on stored object: %v", err)
|
|
}
|
|
etcdW := client.Watch(ctx, key, clientv3.WithRev(int64(rv)))
|
|
|
|
if err := store.Delete(ctx, key, &example.Pod{}, &storage.Preconditions{}, storage.ValidateAllObjectFunc, nil); err != nil {
|
|
t.Fatalf("Delete failed: %v", err)
|
|
}
|
|
|
|
var e watch.Event
|
|
watchCtx, _ := context.WithTimeout(ctx, wait.ForeverTestTimeout)
|
|
select {
|
|
case e = <-w.ResultChan():
|
|
case <-watchCtx.Done():
|
|
t.Fatalf("timed out waiting for watch event")
|
|
}
|
|
deletedRV, err := deletedRevision(watchCtx, etcdW)
|
|
if err != nil {
|
|
t.Fatalf("did not see delete event in raw watch: %v", err)
|
|
}
|
|
watchedDeleteObj := e.Object.(*example.Pod)
|
|
|
|
watchedDeleteRev, err := store.versioner.ParseResourceVersion(watchedDeleteObj.ResourceVersion)
|
|
if err != nil {
|
|
t.Fatalf("ParseWatchResourceVersion failed: %v", err)
|
|
}
|
|
if int64(watchedDeleteRev) != deletedRV {
|
|
t.Errorf("Object from delete event have version: %v, should be the same as etcd delete's mod rev: %d",
|
|
watchedDeleteRev, deletedRV)
|
|
}
|
|
}
|
|
|
|
func deletedRevision(ctx context.Context, watch <-chan clientv3.WatchResponse) (int64, error) {
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return 0, ctx.Err()
|
|
case wres := <-watch:
|
|
for _, evt := range wres.Events {
|
|
if evt.Type == mvccpb.DELETE && evt.Kv != nil {
|
|
return evt.Kv.ModRevision, nil
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestWatchInitializationSignal(t *testing.T) {
|
|
_, store, _ := testSetup(t)
|
|
|
|
ctx, _ := context.WithTimeout(context.Background(), 5*time.Second)
|
|
initSignal := utilflowcontrol.NewInitializationSignal()
|
|
ctx = utilflowcontrol.WithInitializationSignal(ctx, initSignal)
|
|
|
|
key, storedObj := testPropogateStore(ctx, t, store, &example.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo"}})
|
|
_, err := store.Watch(ctx, key, storage.ListOptions{ResourceVersion: storedObj.ResourceVersion, Predicate: storage.Everything})
|
|
if err != nil {
|
|
t.Fatalf("Watch failed: %v", err)
|
|
}
|
|
|
|
initSignal.Wait()
|
|
}
|
|
|
|
func TestProgressNotify(t *testing.T) {
|
|
codec := apitesting.TestCodec(codecs, examplev1.SchemeGroupVersion)
|
|
clusterConfig := testserver.NewTestConfig(t)
|
|
clusterConfig.ExperimentalWatchProgressNotifyInterval = time.Second
|
|
client := testserver.RunEtcd(t, clusterConfig)
|
|
store := newStore(client, codec, newPod, "", schema.GroupResource{Resource: "pods"}, &prefixTransformer{prefix: []byte(defaultTestPrefix)}, false, newTestLeaseManagerConfig())
|
|
ctx := context.Background()
|
|
|
|
key := "/somekey"
|
|
input := &example.Pod{ObjectMeta: metav1.ObjectMeta{Name: "name"}}
|
|
out := &example.Pod{}
|
|
if err := store.Create(ctx, key, input, out, 0); err != nil {
|
|
t.Fatalf("Create failed: %v", err)
|
|
}
|
|
|
|
opts := storage.ListOptions{
|
|
ResourceVersion: out.ResourceVersion,
|
|
Predicate: storage.Everything,
|
|
ProgressNotify: true,
|
|
}
|
|
w, err := store.Watch(ctx, key, opts)
|
|
if err != nil {
|
|
t.Fatalf("Watch failed: %v", err)
|
|
}
|
|
result := &example.Pod{ObjectMeta: metav1.ObjectMeta{ResourceVersion: out.ResourceVersion}}
|
|
testCheckResult(t, watch.Bookmark, w, result)
|
|
}
|
|
|
|
type testWatchStruct struct {
|
|
obj *example.Pod
|
|
expectEvent bool
|
|
watchType watch.EventType
|
|
}
|
|
|
|
type testCodec struct {
|
|
runtime.Codec
|
|
}
|
|
|
|
func (c *testCodec) Decode(data []byte, defaults *schema.GroupVersionKind, into runtime.Object) (runtime.Object, *schema.GroupVersionKind, error) {
|
|
return nil, nil, errTestingDecode
|
|
}
|
|
|
|
func testCheckEventType(t *testing.T, expectEventType watch.EventType, w watch.Interface) {
|
|
select {
|
|
case res := <-w.ResultChan():
|
|
if res.Type != expectEventType {
|
|
t.Errorf("event type want=%v, get=%v", expectEventType, res.Type)
|
|
}
|
|
case <-time.After(wait.ForeverTestTimeout):
|
|
t.Errorf("time out after waiting %v on ResultChan", wait.ForeverTestTimeout)
|
|
}
|
|
}
|
|
|
|
func testCheckResult(t *testing.T, expectEventType watch.EventType, w watch.Interface, expectObj *example.Pod) {
|
|
select {
|
|
case res := <-w.ResultChan():
|
|
if res.Type != expectEventType {
|
|
t.Errorf("event type want=%v, get=%v", expectEventType, res.Type)
|
|
return
|
|
}
|
|
expectNoDiff(t, "incorrect obj", expectObj, res.Object)
|
|
case <-time.After(wait.ForeverTestTimeout):
|
|
t.Errorf("time out after waiting %v on ResultChan", wait.ForeverTestTimeout)
|
|
}
|
|
}
|
|
|
|
func testCheckStop(t *testing.T, w watch.Interface) {
|
|
select {
|
|
case e, ok := <-w.ResultChan():
|
|
if ok {
|
|
var obj string
|
|
switch e.Object.(type) {
|
|
case *example.Pod:
|
|
obj = e.Object.(*example.Pod).Name
|
|
case *metav1.Status:
|
|
obj = e.Object.(*metav1.Status).Message
|
|
}
|
|
t.Errorf("ResultChan should have been closed. Event: %s. Object: %s", e.Type, obj)
|
|
}
|
|
case <-time.After(wait.ForeverTestTimeout):
|
|
t.Errorf("time out after waiting 1s on ResultChan")
|
|
}
|
|
}
|