UPSTREAM: <carry>: sets X-OpenShift-Internal-If-Not-Ready HTTP Header for GC and Namespace controllers

In general, setting the header will result in getting 429 when the server hasn't been ready.
This prevents certain controllers like GC, Namespace from accidentally removing resources when the caches haven't been fully synchronized.

OpenShift-Rebase-Source: 2ebf199

Author: p0lyn0mial

cmd/kube-controller-manager/app/config/patch.go

@@ -15,4 +15,5 @@ type OpenShiftContext struct {
 	UnsupportedKubeAPIOverPreferredHost bool
 	PreferredHostRoundTripperWrapperFn  transport.WrapperFunc
 	PreferredHostHealthMonitor          *health.Prober
+	CustomRoundTrippers                 []transport.WrapperFunc
 }

cmd/kube-controller-manager/app/controllermanager.go

@@ -134,7 +134,7 @@ controller, and serviceaccounts controller.`,
 			}
 			cliflag.PrintFlags(cmd.Flags())
 
-			if err := SetUpPreferredHostForOpenShift(s); err != nil {
+			if err := SetUpCustomRoundTrippersForOpenShift(s); err != nil {
 				fmt.Fprintf(os.Stderr, "%v\n", err)
 				os.Exit(1)
 			}

cmd/kube-controller-manager/app/options/options.go

@@ -502,6 +502,9 @@ func (s KubeControllerManagerOptions) Config(allControllers []string, disabledBy
 		libgorestclient.DefaultServerName(kubeconfig)
 		kubeconfig.Wrap(s.OpenShiftContext.PreferredHostRoundTripperWrapperFn)
 	}
+	for _, customOpenShiftRoundTripper := range s.OpenShiftContext.CustomRoundTrippers {
+		kubeconfig.Wrap(customOpenShiftRoundTripper)
+	}
 
 	client, err := clientset.NewForConfig(restclient.AddUserAgent(kubeconfig, KubeControllerManagerUserAgent))
 	if err != nil {

cmd/kube-controller-manager/app/patch.go

@@ -3,14 +3,17 @@ package app
 import (
 	"fmt"
 	"io/ioutil"
+	"net/http"
 	"path"
+	"strings"
 	"time"
 
 	"k8s.io/apimachinery/pkg/util/json"
 	kyaml "k8s.io/apimachinery/pkg/util/yaml"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/client-go/transport"
 	"k8s.io/component-base/metrics/legacyregistry"
 	"k8s.io/kubernetes/cmd/kube-controller-manager/app/config"
 	"k8s.io/kubernetes/cmd/kube-controller-manager/app/options"
@@ -21,7 +24,9 @@ import (
 
 var InformerFactoryOverride informers.SharedInformerFactory
 
-func SetUpPreferredHostForOpenShift(controllerManagerOptions *options.KubeControllerManagerOptions) error {
+func SetUpCustomRoundTrippersForOpenShift(controllerManagerOptions *options.KubeControllerManagerOptions) error {
+	controllerManagerOptions.OpenShiftContext.CustomRoundTrippers = []transport.WrapperFunc{newRejectIfNotReadyHeaderRoundTripper([]string{"generic-garbage-collector", "namespace-controller"})}
+
 	if !controllerManagerOptions.OpenShiftContext.UnsupportedKubeAPIOverPreferredHost {
 		return nil
 	}
@@ -54,6 +59,7 @@ func SetUpPreferredHostForOpenShift(controllerManagerOptions *options.KubeContro
 
 	controllerManagerOptions.Authentication.WithCustomRoundTripper(controllerManagerOptions.OpenShiftContext.PreferredHostRoundTripperWrapperFn)
 	controllerManagerOptions.Authorization.WithCustomRoundTripper(controllerManagerOptions.OpenShiftContext.PreferredHostRoundTripperWrapperFn)
+
 	return nil
 }
 
@@ -133,3 +139,28 @@ func createRestConfigForHealthMonitor(restConfig *rest.Config) *rest.Config {
 
 	return &restConfigCopy
 }
+
+// newRejectIfNotReadyHeaderRoundTripper a middleware for setting X-OpenShift-Internal-If-Not-Ready HTTP Header for the given users.
+// In general, setting the header will result in getting 429 when the server hasn't been ready.
+// This prevents certain controllers like GC, Namespace from accidentally removing resources when the caches haven't been fully synchronized.
+func newRejectIfNotReadyHeaderRoundTripper(eligibleUsers []string) func(http.RoundTripper) http.RoundTripper {
+	return func(rt http.RoundTripper) http.RoundTripper {
+		return &rejectIfNotReadyHeaderRT{baseRT: rt, eligibleUsers: eligibleUsers}
+	}
+}
+
+type rejectIfNotReadyHeaderRT struct {
+	baseRT        http.RoundTripper
+	eligibleUsers []string
+}
+
+func (rt *rejectIfNotReadyHeaderRT) RoundTrip(r *http.Request) (*http.Response, error) {
+	currentUser := r.UserAgent()
+	for _, eligibleUser := range rt.eligibleUsers {
+		if strings.Contains(currentUser, eligibleUser) {
+			r.Header.Set("X-OpenShift-Internal-If-Not-Ready", "reject")
+			break
+		}
+	}
+	return rt.baseRT.RoundTrip(r)
+}

cmd/kube-controller-manager/app/patch_test.go

@@ -0,0 +1,74 @@
+package app
+
+import (
+	"fmt"
+	"net/http"
+	"net/textproto"
+	"testing"
+)
+
+func TestRejectIfNotReadyHeaderRT(t *testing.T) {
+	scenarios := []struct {
+		name          string
+		eligibleUsers []string
+		currentUser   string
+		expectHeader  bool
+	}{
+		{
+			name:          "scenario 1: happy path",
+			currentUser:   "system:serviceaccount:kube-system:generic-garbage-collector",
+			eligibleUsers: []string{"generic-garbage-collector", "namespace-controller"},
+			expectHeader:  true,
+		},
+		{
+			name:          "scenario 2: ineligible user",
+			currentUser:   "system:serviceaccount:kube-system:service-account-controller",
+			eligibleUsers: []string{"generic-garbage-collector", "namespace-controller"},
+			expectHeader:  false,
+		},
+	}
+
+	for _, scenario := range scenarios {
+		t.Run(scenario.name, func(t *testing.T) {
+			// set up the test
+			fakeRT := fakeRTFunc(func(r *http.Request) (*http.Response, error) {
+				// this is where we validate if the header was set or not
+				headerSet := func() bool {
+					if len(r.Header.Get("X-OpenShift-Internal-If-Not-Ready")) > 0 {
+						return true
+					}
+					return false
+				}()
+				if scenario.expectHeader && !headerSet {
+					return nil, fmt.Errorf("%v header wasn't set", textproto.CanonicalMIMEHeaderKey("X-OpenShift-Internal-If-Not-Ready"))
+				}
+				if !scenario.expectHeader && headerSet {
+					return nil, fmt.Errorf("didn't expect %v header", textproto.CanonicalMIMEHeaderKey("X-OpenShift-Internal-If-Not-Ready"))
+				}
+				if scenario.expectHeader {
+					if value := r.Header.Get("X-OpenShift-Internal-If-Not-Ready"); value != "reject" {
+						return nil, fmt.Errorf("unexpected value %v in the %v header, expected \"reject\"", value, textproto.CanonicalMIMEHeaderKey("X-OpenShift-Internal-If-Not-Ready"))
+					}
+				}
+				return nil, nil
+			})
+			target := newRejectIfNotReadyHeaderRoundTripper(scenario.eligibleUsers)(fakeRT)
+			req, err := http.NewRequest("GET", "", nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+			req.Header.Set("User-Agent", scenario.currentUser)
+
+			// act and validate
+			if _, err := target.RoundTrip(req); err != nil {
+				t.Fatal(err)
+			}
+		})
+	}
+}
+
+type fakeRTFunc func(r *http.Request) (*http.Response, error)
+
+func (rt fakeRTFunc) RoundTrip(r *http.Request) (*http.Response, error) {
+	return rt(r)
+}