Merge pull request #45747 from derekwaynecarr/owner-ref-block-kubelet

Automatic merge from submit-queue (batch tested with PRs 45826, 45747, 45548, 45606, 41766)

OwnerReferencesPermissionEnforcement ignores pods/status

**What this PR does / why we need it**:
It makes the `OwnerReferencesPermissionEnforcement` admission plug-in ignore pods/status operations.  If a cluster has version skew with its master (1.6) and nodes (1.6-N), the kubelet is doing a pods/status operation and that code path is causing an error because ownerRef has new fields unknown to kubelet.

**Release note**:
```release-note
OwnerReferencesPermissionEnforcement admission plugin ignores pods/status.
```

plugin/pkg/admission/gc/gc_admission.go

@@ -33,8 +33,18 @@ import (
 
 func init() {
 	kubeapiserveradmission.Plugins.Register("OwnerReferencesPermissionEnforcement", func(config io.Reader) (admission.Interface, error) {
+		// the pods/status endpoint is ignored by this plugin since old kubelets
+		// corrupt them.  the pod status strategy ensures status updates cannot mutate
+		// ownerRef.
+		whiteList := []whiteListItem{
+			{
+				groupResource: schema.GroupResource{Resource: "pods"},
+				subresource:   "status",
+			},
+		}
 		return &gcPermissionsEnforcement{
-			Handler: admission.NewHandler(admission.Create, admission.Update),
+			Handler:   admission.NewHandler(admission.Create, admission.Update),
+			whiteList: whiteList,
 		}, nil
 	})
 }
@@ -46,9 +56,35 @@ type gcPermissionsEnforcement struct {
 	authorizer authorizer.Authorizer
 
 	restMapper meta.RESTMapper
+
+	// items in this whitelist are ignored upon admission.
+	// any item in this list must protect against ownerRef mutations
+	// via strategy enforcement.
+	whiteList []whiteListItem
+}
+
+// whiteListItem describes an entry in a whitelist ignored by gc permission enforcement.
+type whiteListItem struct {
+	groupResource schema.GroupResource
+	subresource   string
+}
+
+// isWhiteListed returns true if the specified item is in the whitelist.
+func (a *gcPermissionsEnforcement) isWhiteListed(groupResource schema.GroupResource, subresource string) bool {
+	for _, item := range a.whiteList {
+		if item.groupResource == groupResource && item.subresource == subresource {
+			return true
+		}
+	}
+	return false
 }
 
 func (a *gcPermissionsEnforcement) Admit(attributes admission.Attributes) (err error) {
+	// // if the request is in the whitelist, we skip mutation checks for this resource.
+	if a.isWhiteListed(attributes.GetResource().GroupResource(), attributes.GetSubresource()) {
+		return nil
+	}
+
 	// if we aren't changing owner references, then the edit is always allowed
 	if !isChangingOwnerReference(attributes.GetObject(), attributes.GetOldObject()) {
 		return nil

plugin/pkg/admission/gc/gc_admission_test.go

@@ -61,8 +61,18 @@ func (fakeAuthorizer) Authorize(a authorizer.Attributes) (bool, string, error) {
 
 // newGCPermissionsEnforcement returns the admission controller configured for testing.
 func newGCPermissionsEnforcement() *gcPermissionsEnforcement {
+	// the pods/status endpoint is ignored by this plugin since old kubelets
+	// corrupt them.  the pod status strategy ensures status updates cannot mutate
+	// ownerRef.
+	whiteList := []whiteListItem{
+		{
+			groupResource: schema.GroupResource{Resource: "pods"},
+			subresource:   "status",
+		},
+	}
 	gcAdmit := &gcPermissionsEnforcement{
-		Handler: admission.NewHandler(admission.Create, admission.Update),
+		Handler:   admission.NewHandler(admission.Create, admission.Update),
+		whiteList: whiteList,
 	}
 	pluginInitializer := kubeadmission.NewPluginInitializer(nil, nil, fakeAuthorizer{}, nil, api.Registry.RESTMapper())
 	pluginInitializer.Initialize(gcAdmit)
@@ -77,11 +87,12 @@ func TestGCAdmission(t *testing.T) {
 		return strings.Contains(err.Error(), "cannot set an ownerRef on a resource you can't delete")
 	}
 	tests := []struct {
-		name     string
+		name        string
-		username string
+		username    string
-		resource schema.GroupVersionResource
+		resource    schema.GroupVersionResource
-		oldObj   runtime.Object
+		subresource string
-		newObj   runtime.Object
+		oldObj      runtime.Object
+		newObj      runtime.Object
 
 		checkError func(error) bool
 	}{
@@ -199,6 +210,15 @@ func TestGCAdmission(t *testing.T) {
 			newObj:     &api.Pod{},
 			checkError: expectNoError,
 		},
+		{
+			name:        "non-pod-deleter, update status, objectref change",
+			username:    "non-pod-deleter",
+			resource:    api.SchemeGroupVersion.WithResource("pods"),
+			subresource: "status",
+			oldObj:      &api.Pod{},
+			newObj:      &api.Pod{ObjectMeta: metav1.ObjectMeta{OwnerReferences: []metav1.OwnerReference{{Name: "first"}}}},
+			checkError:  expectNoError,
+		},
 		{
 			name:       "non-pod-deleter, update, objectref change",
 			username:   "non-pod-deleter",
@@ -224,7 +244,7 @@ func TestGCAdmission(t *testing.T) {
 			operation = admission.Update
 		}
 		user := &user.DefaultInfo{Name: tc.username}
-		attributes := admission.NewAttributesRecord(tc.newObj, tc.oldObj, schema.GroupVersionKind{}, metav1.NamespaceDefault, "foo", tc.resource, "", operation, user)
+		attributes := admission.NewAttributesRecord(tc.newObj, tc.oldObj, schema.GroupVersionKind{}, metav1.NamespaceDefault, "foo", tc.resource, tc.subresource, operation, user)
 
 		err := gcAdmit.Admit(attributes)
 		if !tc.checkError(err) {
@@ -309,11 +329,12 @@ func TestBlockOwnerDeletionAdmission(t *testing.T) {
 		return strings.Contains(err.Error(), "cannot set blockOwnerDeletion if an ownerReference refers to a resource you can't delete")
 	}
 	tests := []struct {
-		name     string
+		name        string
-		username string
+		username    string
-		resource schema.GroupVersionResource
+		resource    schema.GroupVersionResource
-		oldObj   runtime.Object
+		subresource string
-		newObj   runtime.Object
+		oldObj      runtime.Object
+		newObj      runtime.Object
 
 		checkError func(error) bool
 	}{
@@ -457,7 +478,6 @@ func TestBlockOwnerDeletionAdmission(t *testing.T) {
 			checkError: expectNoError,
 		},
 	}
-
 	gcAdmit := newGCPermissionsEnforcement()
 
 	for _, tc := range tests {
@@ -466,7 +486,7 @@ func TestBlockOwnerDeletionAdmission(t *testing.T) {
 			operation = admission.Update
 		}
 		user := &user.DefaultInfo{Name: tc.username}
-		attributes := admission.NewAttributesRecord(tc.newObj, tc.oldObj, schema.GroupVersionKind{}, metav1.NamespaceDefault, "foo", tc.resource, "", operation, user)
+		attributes := admission.NewAttributesRecord(tc.newObj, tc.oldObj, schema.GroupVersionKind{}, metav1.NamespaceDefault, "foo", tc.resource, tc.subresource, operation, user)
 
 		err := gcAdmit.Admit(attributes)
 		if !tc.checkError(err) {