watch_restart_test.go

Documentation: k8s.io/kubernetes/test/integration/apimachinery

     1  /*
     2  Copyright 2017 The Kubernetes Authors.
     3  
     4  Licensed under the Apache License, Version 2.0 (the "License");
     5  you may not use this file except in compliance with the License.
     6  You may obtain a copy of the License at
     7  
     8      http://www.apache.org/licenses/LICENSE-2.0
     9  
    10  Unless required by applicable law or agreed to in writing, software
    11  distributed under the License is distributed on an "AS IS" BASIS,
    12  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    13  See the License for the specific language governing permissions and
    14  limitations under the License.
    15  */
    16  
    17  package apimachinery
    18  
    19  import (
    20  	"context"
    21  	"fmt"
    22  	"reflect"
    23  	"testing"
    24  	"time"
    25  
    26  	corev1 "k8s.io/api/core/v1"
    27  	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    28  	"k8s.io/apimachinery/pkg/fields"
    29  	"k8s.io/apimachinery/pkg/runtime"
    30  	"k8s.io/apimachinery/pkg/types"
    31  	"k8s.io/apimachinery/pkg/util/wait"
    32  	"k8s.io/apimachinery/pkg/watch"
    33  	"k8s.io/client-go/kubernetes"
    34  	"k8s.io/client-go/tools/cache"
    35  	watchtools "k8s.io/client-go/tools/watch"
    36  	kubeapiservertesting "k8s.io/kubernetes/cmd/kube-apiserver/app/testing"
    37  	"k8s.io/kubernetes/test/integration/framework"
    38  )
    39  
    40  func noopNormalization(output []string) []string {
    41  	return output
    42  }
    43  
    44  func normalizeInformerOutputFunc(initialVal string) func(output []string) []string {
    45  	return func(output []string) []string {
    46  		result := make([]string, 0, len(output))
    47  
    48  		// Removes initial value and all of its direct repetitions
    49  		lastVal := initialVal
    50  		for _, v := range output {
    51  			// Make values unique as informer(List+Watch) duplicates some events
    52  			if v == lastVal {
    53  				continue
    54  			}
    55  			result = append(result, v)
    56  			lastVal = v
    57  		}
    58  
    59  		return result
    60  	}
    61  }
    62  
    63  func noop() {}
    64  
    65  func TestWatchRestartsIfTimeoutNotReached(t *testing.T) {
    66  	// Has to be longer than 5 seconds
    67  	timeout := 30 * time.Second
    68  
    69  	server := kubeapiservertesting.StartTestServerOrDie(t, nil, []string{"--min-request-timeout=7"}, framework.SharedEtcd())
    70  	defer server.TearDownFn()
    71  
    72  	clientset, err := kubernetes.NewForConfig(server.ClientConfig)
    73  	if err != nil {
    74  		t.Fatal(err)
    75  	}
    76  
    77  	namespaceObject := framework.CreateNamespaceOrDie(clientset, "retry-watch", t)
    78  	defer framework.DeleteNamespaceOrDie(clientset, namespaceObject, t)
    79  
    80  	getListFunc := func(c *kubernetes.Clientset, secret *corev1.Secret) func(options metav1.ListOptions) *corev1.SecretList {
    81  		return func(options metav1.ListOptions) *corev1.SecretList {
    82  			options.FieldSelector = fields.OneTermEqualSelector("metadata.name", secret.Name).String()
    83  			res, err := c.CoreV1().Secrets(secret.Namespace).List(context.TODO(), options)
    84  			if err != nil {
    85  				t.Fatalf("Failed to list Secrets: %v", err)
    86  			}
    87  			return res
    88  		}
    89  	}
    90  
    91  	getWatchFunc := func(c *kubernetes.Clientset, secret *corev1.Secret) func(options metav1.ListOptions) (watch.Interface, error) {
    92  		return func(options metav1.ListOptions) (watch.Interface, error) {
    93  			options.FieldSelector = fields.OneTermEqualSelector("metadata.name", secret.Name).String()
    94  			res, err := c.CoreV1().Secrets(secret.Namespace).Watch(context.TODO(), options)
    95  			if err != nil {
    96  				t.Fatalf("Failed to create a watcher on Secrets: %v", err)
    97  			}
    98  			return res, err
    99  		}
   100  	}
   101  
   102  	generateEvents := func(t *testing.T, c *kubernetes.Clientset, secret *corev1.Secret, referenceOutput *[]string, stopChan chan struct{}, stoppedChan chan struct{}) {
   103  		defer close(stoppedChan)
   104  		counter := 0
   105  
   106  		// These 5 seconds are here to protect against a race at the end when we could write something there at the same time as watch.Until ends
   107  		softTimeout := timeout - 5*time.Second
   108  		if softTimeout < 0 {
   109  			panic("Timeout has to be grater than 5 seconds!")
   110  		}
   111  		endChannel := time.After(softTimeout)
   112  		for {
   113  			select {
   114  			// TODO: get this lower once we figure out how to extend ETCD cache
   115  			case <-time.After(1000 * time.Millisecond):
   116  				counter = counter + 1
   117  
   118  				patch := fmt.Sprintf(`{"metadata": {"annotations": {"count": "%d"}}}`, counter)
   119  				_, err := c.CoreV1().Secrets(secret.Namespace).Patch(context.TODO(), secret.Name, types.StrategicMergePatchType, []byte(patch), metav1.PatchOptions{})
   120  				if err != nil {
   121  					t.Errorf("Failed to patch secret: %v", err)
   122  					return
   123  				}
   124  
   125  				*referenceOutput = append(*referenceOutput, fmt.Sprintf("%d", counter))
   126  			case <-endChannel:
   127  				return
   128  			case <-stopChan:
   129  				return
   130  			}
   131  		}
   132  	}
   133  
   134  	initialCount := "0"
   135  	newTestSecret := func(name string) *corev1.Secret {
   136  		return &corev1.Secret{
   137  			ObjectMeta: metav1.ObjectMeta{
   138  				Name:      name,
   139  				Namespace: namespaceObject.Name,
   140  				Annotations: map[string]string{
   141  					"count": initialCount,
   142  				},
   143  			},
   144  			Data: map[string][]byte{
   145  				"data": []byte("value1\n"),
   146  			},
   147  		}
   148  	}
   149  
   150  	tt := []struct {
   151  		name                string
   152  		succeed             bool
   153  		secret              *corev1.Secret
   154  		getWatcher          func(c *kubernetes.Clientset, secret *corev1.Secret) (watch.Interface, error, func())
   155  		normalizeOutputFunc func(referenceOutput []string) []string
   156  	}{
   157  		{
   158  			name:    "regular watcher should fail",
   159  			succeed: false,
   160  			secret:  newTestSecret("secret-01"),
   161  			getWatcher: func(c *kubernetes.Clientset, secret *corev1.Secret) (watch.Interface, error, func()) {
   162  				options := metav1.ListOptions{
   163  					ResourceVersion: secret.ResourceVersion,
   164  				}
   165  				w, err := getWatchFunc(c, secret)(options)
   166  				return w, err, noop
   167  			}, // regular watcher; unfortunately destined to fail
   168  			normalizeOutputFunc: noopNormalization,
   169  		},
   170  		{
   171  			name:    "RetryWatcher survives closed watches",
   172  			succeed: true,
   173  			secret:  newTestSecret("secret-02"),
   174  			getWatcher: func(c *kubernetes.Clientset, secret *corev1.Secret) (watch.Interface, error, func()) {
   175  				lw := &cache.ListWatch{
   176  					WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
   177  						return getWatchFunc(c, secret)(options)
   178  					},
   179  				}
   180  				w, err := watchtools.NewRetryWatcher(secret.ResourceVersion, lw)
   181  				return w, err, func() { <-w.Done() }
   182  			},
   183  			normalizeOutputFunc: noopNormalization,
   184  		},
   185  		{
   186  			name:    "InformerWatcher survives closed watches",
   187  			succeed: true,
   188  			secret:  newTestSecret("secret-03"),
   189  			getWatcher: func(c *kubernetes.Clientset, secret *corev1.Secret) (watch.Interface, error, func()) {
   190  				lw := &cache.ListWatch{
   191  					ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
   192  						return getListFunc(c, secret)(options), nil
   193  					},
   194  					WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
   195  						return getWatchFunc(c, secret)(options)
   196  					},
   197  				}
   198  				// there is an inherent race between a producer (generateEvents) and a consumer (the watcher) that needs to be solved here
   199  				// since the watcher is driven by an informer it is crucial to start producing only after the informer has synced
   200  				// otherwise we might not get all expected events since the informer LIST (or watchelist) and only then WATCHES
   201  				// all events received during the initial LIST (or watchlist) will be seen as a single event (to most recent version of an obj)
   202  				_, informer, w, done := watchtools.NewIndexerInformerWatcher(lw, &corev1.Secret{})
   203  				cache.WaitForCacheSync(context.TODO().Done(), informer.HasSynced)
   204  				return w, nil, func() { <-done }
   205  			},
   206  			normalizeOutputFunc: normalizeInformerOutputFunc(initialCount),
   207  		},
   208  	}
   209  
   210  	t.Run("group", func(t *testing.T) {
   211  		for _, tmptc := range tt {
   212  			tc := tmptc // we need to copy it for parallel runs
   213  			t.Run(tc.name, func(t *testing.T) {
   214  				t.Parallel()
   215  				c, err := kubernetes.NewForConfig(server.ClientConfig)
   216  				if err != nil {
   217  					t.Fatalf("Failed to create clientset: %v", err)
   218  				}
   219  
   220  				secret, err := c.CoreV1().Secrets(tc.secret.Namespace).Create(context.TODO(), tc.secret, metav1.CreateOptions{})
   221  				if err != nil {
   222  					t.Fatalf("Failed to create testing secret %s/%s: %v", tc.secret.Namespace, tc.secret.Name, err)
   223  				}
   224  
   225  				watcher, err, doneFn := tc.getWatcher(c, secret)
   226  				if err != nil {
   227  					t.Fatalf("Failed to create watcher: %v", err)
   228  				}
   229  				defer doneFn()
   230  
   231  				var referenceOutput []string
   232  				var output []string
   233  				stopChan := make(chan struct{})
   234  				stoppedChan := make(chan struct{})
   235  				go generateEvents(t, c, secret, &referenceOutput, stopChan, stoppedChan)
   236  
   237  				// Record current time to be able to asses if the timeout has been reached
   238  				startTime := time.Now()
   239  				ctx, cancel := watchtools.ContextWithOptionalTimeout(context.Background(), timeout)
   240  				defer cancel()
   241  				_, err = watchtools.UntilWithoutRetry(ctx, watcher, func(event watch.Event) (bool, error) {
   242  					s, ok := event.Object.(*corev1.Secret)
   243  					if !ok {
   244  						t.Fatalf("Received an object that is not a Secret: %#v", event.Object)
   245  					}
   246  					output = append(output, s.Annotations["count"])
   247  					// Watch will never end voluntarily
   248  					return false, nil
   249  				})
   250  				watchDuration := time.Since(startTime)
   251  				close(stopChan)
   252  				<-stoppedChan
   253  
   254  				output = tc.normalizeOutputFunc(output)
   255  
   256  				t.Logf("Watch duration: %v; timeout: %v", watchDuration, timeout)
   257  
   258  				if err == nil && !tc.succeed {
   259  					t.Fatalf("Watch should have timed out but it exited without an error!")
   260  				}
   261  
   262  				if err != wait.ErrWaitTimeout && tc.succeed {
   263  					t.Fatalf("Watch exited with error: %v!", err)
   264  				}
   265  
   266  				if watchDuration < timeout && tc.succeed {
   267  					t.Fatalf("Watch should have timed out after %v but it timed out prematurely after %v!", timeout, watchDuration)
   268  				}
   269  
   270  				if watchDuration >= timeout && !tc.succeed {
   271  					t.Fatalf("Watch should have timed out but it succeeded!")
   272  				}
   273  
   274  				if tc.succeed && !reflect.DeepEqual(referenceOutput, output) {
   275  					t.Fatalf("Reference and real output differ! We must have lost some events or read some multiple times!\nRef:  %#v\nReal: %#v", referenceOutput, output)
   276  				}
   277  			})
   278  		}
   279  	})
   280  }
   281
View as plain text