timeout_test.go

Documentation: k8s.io/kubernetes/test/integration/apiserver/admissionwebhook

     1  /*
     2  Copyright 2019 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 admissionwebhook
    18  
    19  import (
    20  	"context"
    21  	"crypto/tls"
    22  	"crypto/x509"
    23  	"encoding/json"
    24  	"fmt"
    25  	"io"
    26  	"net/http"
    27  	"net/http/httptest"
    28  	"sort"
    29  	"strings"
    30  	"sync"
    31  	"testing"
    32  	"time"
    33  
    34  	"k8s.io/api/admission/v1beta1"
    35  	admissionregistrationv1 "k8s.io/api/admissionregistration/v1"
    36  	corev1 "k8s.io/api/core/v1"
    37  	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    38  	"k8s.io/apimachinery/pkg/types"
    39  	"k8s.io/apimachinery/pkg/util/sets"
    40  	"k8s.io/apimachinery/pkg/util/wait"
    41  	clientset "k8s.io/client-go/kubernetes"
    42  	"k8s.io/client-go/rest"
    43  	kubeapiservertesting "k8s.io/kubernetes/cmd/kube-apiserver/app/testing"
    44  	"k8s.io/kubernetes/test/integration/framework"
    45  )
    46  
    47  const (
    48  	testTimeoutClientUsername = "webhook-timeout-integration-client"
    49  )
    50  
    51  // TestWebhookTimeoutWithWatchCache ensures that the admission webhook timeout policy is applied correctly with the watch cache enabled.
    52  func TestWebhookTimeoutWithWatchCache(t *testing.T) {
    53  	testWebhookTimeout(t, true)
    54  }
    55  
    56  // TestWebhookTimeoutWithoutWatchCache ensures that the admission webhook timeout policy is applied correctly without the watch cache enabled.
    57  func TestWebhookTimeoutWithoutWatchCache(t *testing.T) {
    58  	testWebhookTimeout(t, false)
    59  }
    60  
    61  type invocation struct {
    62  	path           string
    63  	timeoutSeconds int
    64  }
    65  
    66  // testWebhookTimeout ensures that the admission webhook timeout policy is applied correctly.
    67  func testWebhookTimeout(t *testing.T, watchCache bool) {
    68  	type testWebhook struct {
    69  		path           string
    70  		timeoutSeconds int32
    71  		policy         admissionregistrationv1.FailurePolicyType
    72  		objectSelector *metav1.LabelSelector
    73  	}
    74  
    75  	testCases := []struct {
    76  		name               string
    77  		timeoutSeconds     int32
    78  		mutatingWebhooks   []testWebhook
    79  		validatingWebhooks []testWebhook
    80  		expectInvocations  []invocation
    81  		expectError        bool
    82  		errorContainsAnyOf []string
    83  	}{
    84  		{
    85  			name:           "minimum of request timeout or webhook timeout propagated",
    86  			timeoutSeconds: 10,
    87  			mutatingWebhooks: []testWebhook{
    88  				{path: "/mutating/1/0s", policy: admissionregistrationv1.Fail, timeoutSeconds: 20},
    89  				{path: "/mutating/2/0s", policy: admissionregistrationv1.Fail, timeoutSeconds: 5},
    90  			},
    91  			validatingWebhooks: []testWebhook{
    92  				{path: "/validating/3/0s", policy: admissionregistrationv1.Fail, timeoutSeconds: 20},
    93  				{path: "/validating/4/0s", policy: admissionregistrationv1.Fail, timeoutSeconds: 5},
    94  			},
    95  			expectInvocations: []invocation{
    96  				{path: "/mutating/1/0s", timeoutSeconds: 10},   // from request
    97  				{path: "/mutating/2/0s", timeoutSeconds: 5},    // from webhook config
    98  				{path: "/validating/3/0s", timeoutSeconds: 10}, // from request
    99  				{path: "/validating/4/0s", timeoutSeconds: 5},  // from webhook config
   100  			},
   101  		},
   102  		{
   103  			name:           "webhooks consume client timeout available, not webhook timeout",
   104  			timeoutSeconds: 10,
   105  			mutatingWebhooks: []testWebhook{
   106  				{path: "/mutating/1/1s", policy: admissionregistrationv1.Fail, timeoutSeconds: 20},
   107  				{path: "/mutating/2/1s", policy: admissionregistrationv1.Fail, timeoutSeconds: 5},
   108  				{path: "/mutating/3/1s", policy: admissionregistrationv1.Fail, timeoutSeconds: 20},
   109  			},
   110  			validatingWebhooks: []testWebhook{
   111  				{path: "/validating/4/1s", policy: admissionregistrationv1.Fail, timeoutSeconds: 5},
   112  				{path: "/validating/5/1s", policy: admissionregistrationv1.Fail, timeoutSeconds: 10},
   113  				{path: "/validating/6/1s", policy: admissionregistrationv1.Fail, timeoutSeconds: 20},
   114  			},
   115  			expectInvocations: []invocation{
   116  				{path: "/mutating/1/1s", timeoutSeconds: 10},  // from request
   117  				{path: "/mutating/2/1s", timeoutSeconds: 5},   // from webhook config (less than request - 1s consumed)
   118  				{path: "/mutating/3/1s", timeoutSeconds: 8},   // from request - 2s consumed
   119  				{path: "/validating/4/1s", timeoutSeconds: 5}, // from webhook config (less than request - 3s consumed by mutating)
   120  				{path: "/validating/5/1s", timeoutSeconds: 7}, // from request - 3s consumed by mutating
   121  				{path: "/validating/6/1s", timeoutSeconds: 7}, // from request - 3s consumed by mutating
   122  			},
   123  		},
   124  		{
   125  			name:           "timed out client requests skip later mutating webhooks (regardless of failure policy) and fail",
   126  			timeoutSeconds: 3,
   127  			mutatingWebhooks: []testWebhook{
   128  				{path: "/mutating/1/5s", policy: admissionregistrationv1.Ignore, timeoutSeconds: 4},
   129  				{path: "/mutating/2/1s", policy: admissionregistrationv1.Ignore, timeoutSeconds: 5},
   130  				{path: "/mutating/3/1s", policy: admissionregistrationv1.Ignore, timeoutSeconds: 5},
   131  			},
   132  			expectInvocations: []invocation{
   133  				{path: "/mutating/1/5s", timeoutSeconds: 3}, // from request
   134  			},
   135  			expectError: true,
   136  			errorContainsAnyOf: []string{
   137  				// refer to https://github.com/kubernetes/kubernetes/issues/98606#issuecomment-774832633
   138  				// for the reason for triggering this scenario
   139  				"stream error",
   140  				"the server was unable to return a response in the time allotted",
   141  			},
   142  		},
   143  	}
   144  
   145  	roots := x509.NewCertPool()
   146  	if !roots.AppendCertsFromPEM(localhostCert) {
   147  		t.Fatal("Failed to append Cert from PEM")
   148  	}
   149  	cert, err := tls.X509KeyPair(localhostCert, localhostKey)
   150  	if err != nil {
   151  		t.Fatalf("Failed to build cert with error: %+v", err)
   152  	}
   153  
   154  	recorder := &timeoutRecorder{invocations: []invocation{}, markers: sets.NewString()}
   155  	webhookServer := httptest.NewUnstartedServer(newTimeoutWebhookHandler(recorder))
   156  	webhookServer.TLS = &tls.Config{
   157  
   158  		RootCAs:      roots,
   159  		Certificates: []tls.Certificate{cert},
   160  	}
   161  	webhookServer.StartTLS()
   162  	defer webhookServer.Close()
   163  
   164  	s := kubeapiservertesting.StartTestServerOrDie(t, kubeapiservertesting.NewDefaultTestServerOptions(), []string{
   165  		"--disable-admission-plugins=ServiceAccount",
   166  		fmt.Sprintf("--watch-cache=%v", watchCache),
   167  	}, framework.SharedEtcd())
   168  	defer s.TearDownFn()
   169  
   170  	// Configure a client with a distinct user name so that it is easy to distinguish requests
   171  	// made by the client from requests made by controllers. We use this to filter out requests
   172  	// before recording them to ensure we don't accidentally mistake requests from controllers
   173  	// as requests made by the client.
   174  	clientConfig := rest.CopyConfig(s.ClientConfig)
   175  	clientConfig.Timeout = 0 // no timeout, we want to set this manually
   176  	clientConfig.Impersonate.UserName = testTimeoutClientUsername
   177  	clientConfig.Impersonate.Groups = []string{"system:masters", "system:authenticated"}
   178  	client, err := clientset.NewForConfig(clientConfig)
   179  	if err != nil {
   180  		t.Fatalf("unexpected error: %v", err)
   181  	}
   182  
   183  	_, err = client.CoreV1().Pods("default").Create(context.TODO(), timeoutMarkerFixture, metav1.CreateOptions{})
   184  	if err != nil {
   185  		t.Fatal(err)
   186  	}
   187  
   188  	for i, tt := range testCases {
   189  		t.Run(tt.name, func(t *testing.T) {
   190  			recorder.Reset()
   191  			ns := fmt.Sprintf("reinvoke-%d", i)
   192  			_, err = client.CoreV1().Namespaces().Create(context.TODO(), &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: ns}}, metav1.CreateOptions{})
   193  			if err != nil {
   194  				t.Fatal(err)
   195  			}
   196  
   197  			mutatingWebhooks := []admissionregistrationv1.MutatingWebhook{}
   198  			for j, webhook := range tt.mutatingWebhooks {
   199  				name := fmt.Sprintf("admission.integration.test.%d.%s", j, strings.Replace(strings.TrimPrefix(webhook.path, "/"), "/", "-", -1))
   200  				endpoint := webhookServer.URL + webhook.path
   201  				mutatingWebhooks = append(mutatingWebhooks, admissionregistrationv1.MutatingWebhook{
   202  					Name: name,
   203  					ClientConfig: admissionregistrationv1.WebhookClientConfig{
   204  						URL:      &endpoint,
   205  						CABundle: localhostCert,
   206  					},
   207  					Rules: []admissionregistrationv1.RuleWithOperations{{
   208  						Operations: []admissionregistrationv1.OperationType{admissionregistrationv1.OperationAll},
   209  						Rule:       admissionregistrationv1.Rule{APIGroups: []string{""}, APIVersions: []string{"v1"}, Resources: []string{"pods"}},
   210  					}},
   211  					ObjectSelector:          webhook.objectSelector,
   212  					FailurePolicy:           &tt.mutatingWebhooks[j].policy,
   213  					TimeoutSeconds:          &tt.mutatingWebhooks[j].timeoutSeconds,
   214  					AdmissionReviewVersions: []string{"v1beta1"},
   215  					SideEffects:             &noSideEffects,
   216  				})
   217  			}
   218  			mutatingCfg, err := client.AdmissionregistrationV1().MutatingWebhookConfigurations().Create(context.TODO(), &admissionregistrationv1.MutatingWebhookConfiguration{
   219  				ObjectMeta: metav1.ObjectMeta{Name: fmt.Sprintf("admission.integration.test-%d", i)},
   220  				Webhooks:   mutatingWebhooks,
   221  			}, metav1.CreateOptions{})
   222  			if err != nil {
   223  				t.Fatal(err)
   224  			}
   225  			defer func() {
   226  				err := client.AdmissionregistrationV1().MutatingWebhookConfigurations().Delete(context.TODO(), mutatingCfg.GetName(), metav1.DeleteOptions{})
   227  				if err != nil {
   228  					t.Fatal(err)
   229  				}
   230  			}()
   231  
   232  			validatingWebhooks := []admissionregistrationv1.ValidatingWebhook{}
   233  			for j, webhook := range tt.validatingWebhooks {
   234  				name := fmt.Sprintf("admission.integration.test.%d.%s", j, strings.Replace(strings.TrimPrefix(webhook.path, "/"), "/", "-", -1))
   235  				endpoint := webhookServer.URL + webhook.path
   236  				validatingWebhooks = append(validatingWebhooks, admissionregistrationv1.ValidatingWebhook{
   237  					Name: name,
   238  					ClientConfig: admissionregistrationv1.WebhookClientConfig{
   239  						URL:      &endpoint,
   240  						CABundle: localhostCert,
   241  					},
   242  					Rules: []admissionregistrationv1.RuleWithOperations{{
   243  						Operations: []admissionregistrationv1.OperationType{admissionregistrationv1.OperationAll},
   244  						Rule:       admissionregistrationv1.Rule{APIGroups: []string{""}, APIVersions: []string{"v1"}, Resources: []string{"pods"}},
   245  					}},
   246  					ObjectSelector:          webhook.objectSelector,
   247  					FailurePolicy:           &tt.validatingWebhooks[j].policy,
   248  					TimeoutSeconds:          &tt.validatingWebhooks[j].timeoutSeconds,
   249  					AdmissionReviewVersions: []string{"v1beta1"},
   250  					SideEffects:             &noSideEffects,
   251  				})
   252  			}
   253  			validatingCfg, err := client.AdmissionregistrationV1().ValidatingWebhookConfigurations().Create(context.TODO(), &admissionregistrationv1.ValidatingWebhookConfiguration{
   254  				ObjectMeta: metav1.ObjectMeta{Name: fmt.Sprintf("admission.integration.test-%d", i)},
   255  				Webhooks:   validatingWebhooks,
   256  			}, metav1.CreateOptions{})
   257  			if err != nil {
   258  				t.Fatal(err)
   259  			}
   260  			defer func() {
   261  				err := client.AdmissionregistrationV1().ValidatingWebhookConfigurations().Delete(context.TODO(), validatingCfg.GetName(), metav1.DeleteOptions{})
   262  				if err != nil {
   263  					t.Fatal(err)
   264  				}
   265  			}()
   266  
   267  			// wait until new webhook is called the first time
   268  			if err := wait.PollImmediate(time.Millisecond*5, wait.ForeverTestTimeout, func() (bool, error) {
   269  				_, err = client.CoreV1().Pods("default").Patch(context.TODO(), timeoutMarkerFixture.Name, types.JSONPatchType, []byte("[]"), metav1.PatchOptions{})
   270  				received := recorder.MarkerReceived()
   271  				if len(tt.mutatingWebhooks) > 0 && !received.Has("mutating") {
   272  					t.Logf("Waiting for mutating webhooks to become effective, getting marker object: %v", err)
   273  					return false, nil
   274  				}
   275  				if len(tt.validatingWebhooks) > 0 && !received.Has("validating") {
   276  					t.Logf("Waiting for validating webhooks to become effective, getting marker object: %v", err)
   277  					return false, nil
   278  				}
   279  				return true, nil
   280  			}); err != nil {
   281  				t.Fatal(err)
   282  			}
   283  
   284  			pod := &corev1.Pod{
   285  				TypeMeta: metav1.TypeMeta{APIVersion: "v1", Kind: "Pod"},
   286  				ObjectMeta: metav1.ObjectMeta{
   287  					Namespace: ns,
   288  					Name:      "labeled",
   289  					Labels:    map[string]string{"x": "true"},
   290  				},
   291  				Spec: corev1.PodSpec{
   292  					Containers: []corev1.Container{{
   293  						Name:  "fake-name",
   294  						Image: "fakeimage",
   295  					}},
   296  				},
   297  			}
   298  
   299  			body, err := json.Marshal(pod)
   300  			if err != nil {
   301  				t.Fatal(err)
   302  			}
   303  
   304  			// set the timeout parameter manually so we don't actually cut off the request client-side, and wait for the server response
   305  			err = client.CoreV1().RESTClient().Post().Resource("pods").Namespace(ns).Body(body).Param("timeout", fmt.Sprintf("%ds", tt.timeoutSeconds)).Do(context.TODO()).Error()
   306  			// _, err = testClient.CoreV1().Pods(ns).Create(pod)
   307  
   308  			if tt.expectError {
   309  				if err == nil {
   310  					t.Fatalf("expected error but got none")
   311  				}
   312  
   313  				expected := false
   314  				if len(tt.errorContainsAnyOf) != 0 {
   315  					for _, errStr := range tt.errorContainsAnyOf {
   316  						if strings.Contains(err.Error(), errStr) {
   317  							expected = true
   318  							break
   319  						}
   320  					}
   321  				}
   322  				if !expected {
   323  					t.Errorf("expected the error to be any of %q, but got: %v", tt.errorContainsAnyOf, err)
   324  				}
   325  				return
   326  			}
   327  
   328  			if err != nil {
   329  				t.Fatal(err)
   330  			}
   331  
   332  			if tt.expectInvocations != nil {
   333  				for i, invocation := range tt.expectInvocations {
   334  					if len(recorder.invocations) <= i {
   335  						t.Errorf("expected invocation of %s, got none", invocation.path)
   336  						continue
   337  					}
   338  
   339  					if recorder.invocations[i].path != invocation.path {
   340  						t.Errorf("expected invocation of %s, got %s", invocation.path, recorder.invocations[i].path)
   341  						continue
   342  					}
   343  					if recorder.invocations[i].timeoutSeconds != invocation.timeoutSeconds {
   344  						t.Errorf("expected invocation of %s with timeout %d, got %d", invocation.path, invocation.timeoutSeconds, recorder.invocations[i].timeoutSeconds)
   345  						continue
   346  					}
   347  				}
   348  
   349  				if len(recorder.invocations) > len(tt.expectInvocations) {
   350  					for _, invocation := range recorder.invocations[len(tt.expectInvocations):] {
   351  						t.Errorf("unexpected invocation of %s", invocation.path)
   352  					}
   353  				}
   354  			}
   355  		})
   356  	}
   357  }
   358  
   359  type timeoutRecorder struct {
   360  	mu          sync.Mutex
   361  	markers     sets.String
   362  	invocations []invocation
   363  }
   364  
   365  // Reset zeros out all counts
   366  func (i *timeoutRecorder) Reset() {
   367  	i.mu.Lock()
   368  	defer i.mu.Unlock()
   369  	i.invocations = []invocation{}
   370  	i.markers = sets.NewString()
   371  }
   372  
   373  // MarkerReceived records the specified markers were received and returns the set of received markers
   374  func (i *timeoutRecorder) MarkerReceived(markers ...string) sets.String {
   375  	i.mu.Lock()
   376  	defer i.mu.Unlock()
   377  	i.markers.Insert(markers...)
   378  	return i.markers.Union(nil)
   379  }
   380  
   381  func (i *timeoutRecorder) RecordInvocation(call invocation) {
   382  	i.mu.Lock()
   383  	defer i.mu.Unlock()
   384  	i.invocations = append(i.invocations, call)
   385  	sort.SliceStable(i.invocations, func(a, b int) bool {
   386  		aValidating := strings.Contains(i.invocations[a].path, "validating")
   387  		bValidating := strings.Contains(i.invocations[b].path, "validating")
   388  		switch {
   389  		case aValidating && bValidating:
   390  			// sort validating by path
   391  			return strings.Compare(i.invocations[a].path, i.invocations[b].path) < 0
   392  		case !aValidating && !bValidating:
   393  			// keep mutating in original order
   394  			return a < b
   395  		case aValidating && !bValidating:
   396  			// put validating last
   397  			return false
   398  		default:
   399  			return true
   400  		}
   401  	})
   402  }
   403  
   404  func newTimeoutWebhookHandler(recorder *timeoutRecorder) http.Handler {
   405  	allow := func(w http.ResponseWriter) {
   406  		w.Header().Set("Content-Type", "application/json")
   407  		json.NewEncoder(w).Encode(&v1beta1.AdmissionReview{
   408  			Response: &v1beta1.AdmissionResponse{
   409  				Allowed: true,
   410  			},
   411  		})
   412  	}
   413  	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
   414  		defer r.Body.Close()
   415  		data, err := io.ReadAll(r.Body)
   416  		if err != nil {
   417  			http.Error(w, err.Error(), 400)
   418  		}
   419  		review := v1beta1.AdmissionReview{}
   420  		if err := json.Unmarshal(data, &review); err != nil {
   421  			http.Error(w, err.Error(), 400)
   422  		}
   423  		if review.Request.UserInfo.Username != testTimeoutClientUsername {
   424  			// skip requests not originating from this integration test's client
   425  			allow(w)
   426  			return
   427  		}
   428  
   429  		if len(review.Request.Object.Raw) == 0 {
   430  			http.Error(w, err.Error(), 400)
   431  		}
   432  		pod := &corev1.Pod{}
   433  		if err := json.Unmarshal(review.Request.Object.Raw, pod); err != nil {
   434  			http.Error(w, err.Error(), 400)
   435  		}
   436  
   437  		// When resetting between tests, a marker object is patched until this webhook
   438  		// observes it, at which point it is considered ready.
   439  		if pod.Namespace == timeoutMarkerFixture.Namespace && pod.Name == timeoutMarkerFixture.Name {
   440  			if strings.HasPrefix(r.URL.Path, "/mutating/") {
   441  				recorder.MarkerReceived("mutating")
   442  			}
   443  			if strings.HasPrefix(r.URL.Path, "/validating/") {
   444  				recorder.MarkerReceived("validating")
   445  			}
   446  			allow(w)
   447  			return
   448  		}
   449  
   450  		timeout, err := time.ParseDuration(r.URL.Query().Get("timeout"))
   451  		if err != nil {
   452  			http.Error(w, err.Error(), http.StatusBadRequest)
   453  		}
   454  		invocation := invocation{path: r.URL.Path, timeoutSeconds: int(timeout.Round(time.Second) / time.Second)}
   455  		recorder.RecordInvocation(invocation)
   456  
   457  		switch {
   458  		case strings.HasSuffix(r.URL.Path, "/0s"):
   459  			allow(w)
   460  		case strings.HasSuffix(r.URL.Path, "/1s"):
   461  			time.Sleep(time.Second)
   462  			allow(w)
   463  		case strings.HasSuffix(r.URL.Path, "/5s"):
   464  			time.Sleep(5 * time.Second)
   465  			allow(w)
   466  		default:
   467  			http.NotFound(w, r)
   468  		}
   469  	})
   470  }
   471  
   472  var timeoutMarkerFixture = &corev1.Pod{
   473  	ObjectMeta: metav1.ObjectMeta{
   474  		Namespace: "default",
   475  		Name:      "marker",
   476  	},
   477  	Spec: corev1.PodSpec{
   478  		Containers: []corev1.Container{{
   479  			Name:  "fake-name",
   480  			Image: "fakeimage",
   481  		}},
   482  	},
   483  }
   484
View as plain text