mirror of
https://github.com/kubernetes/client-go.git
synced 2025-07-14 23:44:26 +00:00
57bc261e52
The ability to automatically stop on context cancellation was new functionality that adds complexity and wasn't really used in Kubernetes. If someone wants this, they can add it outside of the function. A *WithLogger variant avoids the complexity and is consistent with NewStreamWatcherWithLogger over in apimachinery. Kubernetes-commit: 1a8d8c9b4a33daf9330434e1ad544ef3571722a3
467 lines
12 KiB
Go
467 lines
12 KiB
Go
/*
|
|
Copyright 2017 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 watch
|
|
|
|
import (
|
|
"context"
|
|
"reflect"
|
|
goruntime "runtime"
|
|
"sort"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/google/go-cmp/cmp"
|
|
|
|
corev1 "k8s.io/api/core/v1"
|
|
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/apimachinery/pkg/runtime"
|
|
"k8s.io/apimachinery/pkg/runtime/schema"
|
|
"k8s.io/apimachinery/pkg/util/dump"
|
|
"k8s.io/apimachinery/pkg/util/wait"
|
|
"k8s.io/apimachinery/pkg/watch"
|
|
clientfeatures "k8s.io/client-go/features"
|
|
clientfeaturestesting "k8s.io/client-go/features/testing"
|
|
fakeclientset "k8s.io/client-go/kubernetes/fake"
|
|
testcore "k8s.io/client-go/testing"
|
|
"k8s.io/client-go/tools/cache"
|
|
)
|
|
|
|
// TestEventProcessorExit is expected to timeout if the event processor fails
|
|
// to exit when stopped.
|
|
func TestEventProcessorExit(t *testing.T) {
|
|
event := watch.Event{}
|
|
|
|
tests := []struct {
|
|
name string
|
|
write func(e *eventProcessor)
|
|
}{
|
|
{
|
|
name: "exit on blocked read",
|
|
write: func(e *eventProcessor) {
|
|
e.push(event)
|
|
},
|
|
},
|
|
{
|
|
name: "exit on blocked write",
|
|
write: func(e *eventProcessor) {
|
|
e.push(event)
|
|
e.push(event)
|
|
},
|
|
},
|
|
}
|
|
for _, test := range tests {
|
|
t.Run(test.name, func(t *testing.T) {
|
|
out := make(chan watch.Event)
|
|
e := newEventProcessor(out)
|
|
|
|
test.write(e)
|
|
|
|
exited := make(chan struct{})
|
|
go func() {
|
|
e.run()
|
|
close(exited)
|
|
}()
|
|
|
|
<-out
|
|
e.stop()
|
|
goruntime.Gosched()
|
|
<-exited
|
|
})
|
|
}
|
|
}
|
|
|
|
type apiInt int
|
|
|
|
func (apiInt) GetObjectKind() schema.ObjectKind { return nil }
|
|
func (apiInt) DeepCopyObject() runtime.Object { return nil }
|
|
|
|
func TestEventProcessorOrdersEvents(t *testing.T) {
|
|
out := make(chan watch.Event)
|
|
e := newEventProcessor(out)
|
|
go e.run()
|
|
|
|
numProcessed := 0
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
go func() {
|
|
for i := 0; i < 1000; i++ {
|
|
e := <-out
|
|
if got, want := int(e.Object.(apiInt)), i; got != want {
|
|
t.Errorf("unexpected event: got=%d, want=%d", got, want)
|
|
}
|
|
numProcessed++
|
|
}
|
|
cancel()
|
|
}()
|
|
|
|
for i := 0; i < 1000; i++ {
|
|
e.push(watch.Event{Object: apiInt(i)})
|
|
}
|
|
|
|
<-ctx.Done()
|
|
e.stop()
|
|
|
|
if numProcessed != 1000 {
|
|
t.Errorf("unexpected number of events processed: %d", numProcessed)
|
|
}
|
|
|
|
}
|
|
|
|
type byEventTypeAndName []watch.Event
|
|
|
|
func (a byEventTypeAndName) Len() int { return len(a) }
|
|
func (a byEventTypeAndName) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
|
|
func (a byEventTypeAndName) Less(i, j int) bool {
|
|
if a[i].Type < a[j].Type {
|
|
return true
|
|
}
|
|
|
|
if a[i].Type > a[j].Type {
|
|
return false
|
|
}
|
|
|
|
return a[i].Object.(*corev1.Secret).Name < a[j].Object.(*corev1.Secret).Name
|
|
}
|
|
|
|
func newTestSecret(name, key, value string) *corev1.Secret {
|
|
return &corev1.Secret{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: name,
|
|
},
|
|
StringData: map[string]string{
|
|
key: value,
|
|
},
|
|
}
|
|
}
|
|
|
|
func TestNewInformerWatcher(t *testing.T) {
|
|
// Make sure there are no 2 same types of events on a secret with the same name or that might be flaky.
|
|
tt := []struct {
|
|
name string
|
|
watchListFeatureEnabled bool
|
|
objects []runtime.Object
|
|
inputEvents []watch.Event
|
|
outputEvents []watch.Event
|
|
}{
|
|
{
|
|
name: "WatchListClient feature disabled",
|
|
watchListFeatureEnabled: false,
|
|
objects: []runtime.Object{
|
|
newTestSecret("pod-1", "foo-1", "initial"),
|
|
newTestSecret("pod-2", "foo-2", "initial"),
|
|
newTestSecret("pod-3", "foo-3", "initial"),
|
|
},
|
|
inputEvents: []watch.Event{
|
|
{
|
|
Type: watch.Added,
|
|
Object: newTestSecret("pod-4", "foo-4", "initial"),
|
|
},
|
|
{
|
|
Type: watch.Modified,
|
|
Object: newTestSecret("pod-2", "foo-2", "new"),
|
|
},
|
|
{
|
|
Type: watch.Deleted,
|
|
Object: newTestSecret("pod-3", "foo-3", "initial"),
|
|
},
|
|
},
|
|
outputEvents: []watch.Event{
|
|
// When WatchListClient is disabled, ListAndWatch creates fake
|
|
// ADDED events for each object listed.
|
|
{
|
|
Type: watch.Added,
|
|
Object: newTestSecret("pod-1", "foo-1", "initial"),
|
|
},
|
|
{
|
|
Type: watch.Added,
|
|
Object: newTestSecret("pod-2", "foo-2", "initial"),
|
|
},
|
|
{
|
|
Type: watch.Added,
|
|
Object: newTestSecret("pod-3", "foo-3", "initial"),
|
|
},
|
|
// Normal events follow.
|
|
{
|
|
Type: watch.Added,
|
|
Object: newTestSecret("pod-4", "foo-4", "initial"),
|
|
},
|
|
{
|
|
Type: watch.Modified,
|
|
Object: newTestSecret("pod-2", "foo-2", "new"),
|
|
},
|
|
{
|
|
Type: watch.Deleted,
|
|
Object: newTestSecret("pod-3", "foo-3", "initial"),
|
|
},
|
|
},
|
|
},
|
|
{
|
|
name: "WatchListClient feature enabled",
|
|
watchListFeatureEnabled: true,
|
|
objects: []runtime.Object{
|
|
newTestSecret("pod-1", "foo-1", "initial"),
|
|
newTestSecret("pod-2", "foo-2", "initial"),
|
|
newTestSecret("pod-3", "foo-3", "initial"),
|
|
},
|
|
inputEvents: []watch.Event{
|
|
{
|
|
Type: watch.Added,
|
|
Object: newTestSecret("pod-1", "foo-1", "initial"),
|
|
},
|
|
{
|
|
Type: watch.Added,
|
|
Object: newTestSecret("pod-2", "foo-2", "initial"),
|
|
},
|
|
{
|
|
Type: watch.Added,
|
|
Object: newTestSecret("pod-3", "foo-3", "initial"),
|
|
},
|
|
// ListWatch bookmark indicates that initial listing is done
|
|
{
|
|
Type: watch.Bookmark,
|
|
Object: &corev1.Secret{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Annotations: map[string]string{
|
|
metav1.InitialEventsAnnotationKey: "true",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
Type: watch.Added,
|
|
Object: newTestSecret("pod-4", "foo-4", "initial"),
|
|
},
|
|
{
|
|
Type: watch.Modified,
|
|
Object: newTestSecret("pod-2", "foo-2", "new"),
|
|
},
|
|
{
|
|
Type: watch.Deleted,
|
|
Object: newTestSecret("pod-3", "foo-3", "initial"),
|
|
},
|
|
},
|
|
outputEvents: []watch.Event{
|
|
// When WatchListClient is enabled, WatchList receives
|
|
// ADDED events from the server for each existing object.
|
|
{
|
|
Type: watch.Added,
|
|
Object: newTestSecret("pod-1", "foo-1", "initial"),
|
|
},
|
|
{
|
|
Type: watch.Added,
|
|
Object: newTestSecret("pod-2", "foo-2", "initial"),
|
|
},
|
|
{
|
|
Type: watch.Added,
|
|
Object: newTestSecret("pod-3", "foo-3", "initial"),
|
|
},
|
|
// Bookmark event at the end of listing is not sent to the client.
|
|
// Normal events follow.
|
|
{
|
|
Type: watch.Added,
|
|
Object: newTestSecret("pod-4", "foo-4", "initial"),
|
|
},
|
|
{
|
|
Type: watch.Modified,
|
|
Object: newTestSecret("pod-2", "foo-2", "new"),
|
|
},
|
|
{
|
|
Type: watch.Deleted,
|
|
Object: newTestSecret("pod-3", "foo-3", "initial"),
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
for _, tc := range tt {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
clientfeaturestesting.SetFeatureDuringTest(t, clientfeatures.WatchListClient, tc.watchListFeatureEnabled)
|
|
|
|
fake := fakeclientset.NewSimpleClientset(tc.objects...)
|
|
inputCh := make(chan watch.Event)
|
|
inputWatcher := watch.NewProxyWatcher(inputCh)
|
|
// Indexer should stop the input watcher when the output watcher is stopped.
|
|
// But stop it at the end of the test, just in case.
|
|
defer inputWatcher.Stop()
|
|
inputStopCh := inputWatcher.StopChan()
|
|
fake.PrependWatchReactor("secrets", testcore.DefaultWatchReactor(inputWatcher, nil))
|
|
// Send events and then close the done channel
|
|
inputDoneCh := make(chan struct{})
|
|
go func() {
|
|
defer close(inputDoneCh)
|
|
for _, e := range tc.inputEvents {
|
|
select {
|
|
case inputCh <- e:
|
|
case <-inputStopCh:
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
|
|
lw := &cache.ListWatch{
|
|
ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
|
|
return fake.CoreV1().Secrets("").List(context.TODO(), options)
|
|
},
|
|
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
|
|
return fake.CoreV1().Secrets("").Watch(context.TODO(), options)
|
|
},
|
|
}
|
|
//nolint:logcheck // Intentionally uses the older API.
|
|
_, _, outputWatcher, informerDoneCh := NewIndexerInformerWatcher(lw, &corev1.Secret{})
|
|
outputCh := outputWatcher.ResultChan()
|
|
timeoutCh := time.After(wait.ForeverTestTimeout)
|
|
var result []watch.Event
|
|
loop:
|
|
for {
|
|
select {
|
|
case event, ok := <-outputCh:
|
|
if !ok {
|
|
t.Errorf("Output result channel closed prematurely")
|
|
break loop
|
|
}
|
|
result = append(result, *event.DeepCopy())
|
|
if len(result) >= len(tc.outputEvents) {
|
|
break loop
|
|
}
|
|
case <-timeoutCh:
|
|
t.Error("Timed out waiting for events")
|
|
break loop
|
|
}
|
|
}
|
|
|
|
// Informers don't guarantee event order so we need to sort these arrays to compare them.
|
|
sort.Sort(byEventTypeAndName(tc.outputEvents))
|
|
sort.Sort(byEventTypeAndName(result))
|
|
|
|
if !reflect.DeepEqual(tc.outputEvents, result) {
|
|
t.Errorf("\nexpected: %s,\ngot: %s,\ndiff: %s", dump.Pretty(tc.outputEvents), dump.Pretty(result), cmp.Diff(tc.outputEvents, result))
|
|
return
|
|
}
|
|
|
|
// Send some more events, but don't read them.
|
|
// Stop producing events when the consumer stops the watcher.
|
|
go func() {
|
|
defer close(inputCh)
|
|
for _, e := range tc.inputEvents {
|
|
select {
|
|
case inputCh <- e:
|
|
case <-inputStopCh:
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
|
|
// Stop before reading all the data to make sure the informer can deal with closed channel
|
|
outputWatcher.Stop()
|
|
|
|
select {
|
|
case <-informerDoneCh:
|
|
case <-time.After(wait.ForeverTestTimeout):
|
|
t.Error("Timed out waiting for informer to cleanup")
|
|
}
|
|
})
|
|
}
|
|
|
|
}
|
|
|
|
// TestInformerWatcherDeletedFinalStateUnknown tests the code path when `DeleteFunc`
|
|
// in `NewIndexerInformerWatcher` receives a `cache.DeletedFinalStateUnknown`
|
|
// object from the underlying `DeltaFIFO`. The triggering condition is described
|
|
// at https://github.com/kubernetes/kubernetes/blob/dc39ab2417bfddcec37be4011131c59921fdbe98/staging/src/k8s.io/client-go/tools/cache/delta_fifo.go#L736-L739.
|
|
//
|
|
// Code from @liggitt
|
|
func TestInformerWatcherDeletedFinalStateUnknown(t *testing.T) {
|
|
listCalls := 0
|
|
watchCalls := 0
|
|
lw := &cache.ListWatch{
|
|
ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
|
|
retval := &corev1.SecretList{}
|
|
if listCalls == 0 {
|
|
// Return a list with items in it
|
|
retval.ResourceVersion = "1"
|
|
retval.Items = []corev1.Secret{{ObjectMeta: metav1.ObjectMeta{Name: "secret1", Namespace: "ns1", ResourceVersion: "123"}}}
|
|
} else {
|
|
// Return empty lists after the first call
|
|
retval.ResourceVersion = "2"
|
|
}
|
|
listCalls++
|
|
return retval, nil
|
|
},
|
|
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
|
|
w := watch.NewRaceFreeFake()
|
|
if options.ResourceVersion == "1" {
|
|
go func() {
|
|
// Close with a "Gone" error when trying to start a watch from the first list
|
|
w.Error(&apierrors.NewGone("gone").ErrStatus)
|
|
w.Stop()
|
|
}()
|
|
}
|
|
watchCalls++
|
|
return w, nil
|
|
},
|
|
}
|
|
//nolint:logcheck // Intentionally uses the older API.
|
|
_, _, w, done := NewIndexerInformerWatcher(lw, &corev1.Secret{})
|
|
defer w.Stop()
|
|
|
|
// Expect secret add
|
|
select {
|
|
case event, ok := <-w.ResultChan():
|
|
if !ok {
|
|
t.Fatal("unexpected close")
|
|
}
|
|
if event.Type != watch.Added {
|
|
t.Fatalf("expected Added event, got %#v", event)
|
|
}
|
|
if event.Object.(*corev1.Secret).ResourceVersion != "123" {
|
|
t.Fatalf("expected added Secret with rv=123, got %#v", event.Object)
|
|
}
|
|
case <-time.After(time.Second * 10):
|
|
t.Fatal("timeout")
|
|
}
|
|
|
|
// Expect secret delete because the relist was missing the secret
|
|
select {
|
|
case event, ok := <-w.ResultChan():
|
|
if !ok {
|
|
t.Fatal("unexpected close")
|
|
}
|
|
if event.Type != watch.Deleted {
|
|
t.Fatalf("expected Deleted event, got %#v", event)
|
|
}
|
|
if event.Object.(*corev1.Secret).ResourceVersion != "123" {
|
|
t.Fatalf("expected deleted Secret with rv=123, got %#v", event.Object)
|
|
}
|
|
case <-time.After(time.Second * 10):
|
|
t.Fatal("timeout")
|
|
}
|
|
|
|
w.Stop()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(time.Second * 10):
|
|
t.Fatal("timeout")
|
|
}
|
|
|
|
if listCalls < 2 {
|
|
t.Fatalf("expected at least 2 list calls, got %d", listCalls)
|
|
}
|
|
if watchCalls < 1 {
|
|
t.Fatalf("expected at least 1 watch call, got %d", watchCalls)
|
|
}
|
|
}
|