Merge pull request #51082 from caesarxuchao/repair-null-pending-initializer

Automatic merge from submit-queue (batch tested with PRs 50953, 51082)

Fix mergekey of initializers; Repair invalid update of initializers

Fix https://github.com/kubernetes/kubernetes/issues/51131

The PR did two things to make parallel patching `metadata.initializers.pending` possible:
* Add mergekey to initializers.pending
* Let the initializer admission plugin set the `metadata.intializers` to nil if an update makes the `pending` and the `result` both nil, instead of returning a validation error. Otherwise if multiple initializer controllers sending the patch removing themselves from `pending` at the same time, one of them will get a validation error.


```release-note
The patch to remove the last initializer from metadata.initializer.pending will result in metadata.initializer to be set to nil (assuming metadata.initializer.result is also nil), instead of resulting in an validation error.
```

api/openapi-spec/swagger.json

@@ -65826,7 +65826,9 @@
       "type": "array",
       "items": {
        "$ref": "#/definitions/io.k8s.apimachinery.pkg.apis.meta.v1.Initializer"
-      }
+      },
+      "x-kubernetes-patch-merge-key": "name",
+      "x-kubernetes-patch-strategy": "merge"
      },
      "result": {
       "description": "If result is set with the Failure field, the object will be persisted to storage and then deleted, ensuring that other clients can observe the deletion.",

federation/apis/openapi-spec/swagger.json

@@ -13441,7 +13441,9 @@
       "type": "array",
       "items": {
        "$ref": "#/definitions/io.k8s.apimachinery.pkg.apis.meta.v1.Initializer"
-      }
+      },
+      "x-kubernetes-patch-merge-key": "name",
+      "x-kubernetes-patch-strategy": "merge"
      },
      "result": {
       "description": "If result is set with the Failure field, the object will be persisted to storage and then deleted, ensuring that other clients can observe the deletion.",

plugin/pkg/admission/initialization/BUILD

@@ -46,6 +46,12 @@ go_test(
     library = ":go_default_library",
     deps = [
         "//vendor/k8s.io/api/admissionregistration/v1alpha1:go_default_library",
+        "//vendor/k8s.io/api/core/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/api/meta:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/admission:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/authorization/authorizer:go_default_library",
     ],
 )

plugin/pkg/admission/initialization/initialization.go

@@ -208,6 +208,9 @@ func (i *initializer) Admit(a admission.Attributes) (err error) {
 
 		glog.V(5).Infof("Modifying uninitialized resource %s", a.GetResource())
 
+		if updated != nil && len(updated.Pending) == 0 && updated.Result == nil {
+			accessor.SetInitializers(nil)
+		}
 		// because we are called before validation, we need to ensure the update transition is valid.
 		if errs := validation.ValidateInitializersUpdate(updated, existing, initializerFieldPath); len(errs) > 0 {
 			return errors.NewInvalid(a.GetKind().GroupKind(), a.GetName(), errs)

plugin/pkg/admission/initialization/initialization_test.go

@@ -18,10 +18,17 @@ package initialization
 
 import (
 	"reflect"
+	"strings"
 	"testing"
 
 	"k8s.io/api/admissionregistration/v1alpha1"
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apiserver/pkg/admission"
+	"k8s.io/apiserver/pkg/authorization/authorizer"
 )
 
 func newInitializer(name string, rules ...v1alpha1.Rule) *v1alpha1.InitializerConfiguration {
@@ -97,3 +104,75 @@ func TestFindInitializers(t *testing.T) {
 		})
 	}
 }
+
+type fakeAuthorizer struct {
+	accept bool
+}
+
+func (f *fakeAuthorizer) Authorize(a authorizer.Attributes) (bool, string, error) {
+	if f.accept {
+		return true, "", nil
+	}
+	return false, "denied", nil
+}
+
+func TestAdmitUpdate(t *testing.T) {
+	tests := []struct {
+		name             string
+		oldInitializers  *metav1.Initializers
+		newInitializers  *metav1.Initializers
+		verifyUpdatedObj func(runtime.Object) (pass bool, reason string)
+		err              string
+	}{
+		{
+			name:            "updates on initialized resources are allowed",
+			oldInitializers: nil,
+			newInitializers: nil,
+			err:             "",
+		},
+		{
+			name:            "updates on initialized resources are allowed",
+			oldInitializers: &metav1.Initializers{Pending: []metav1.Initializer{{Name: "init.k8s.io"}}},
+			newInitializers: &metav1.Initializers{},
+			verifyUpdatedObj: func(obj runtime.Object) (bool, string) {
+				accessor, err := meta.Accessor(obj)
+				if err != nil {
+					return false, "cannot get accessor"
+				}
+				if accessor.GetInitializers() != nil {
+					return false, "expect nil initializers"
+				}
+				return true, ""
+			},
+			err: "",
+		},
+		{
+			name:            "initializers may not be set once initialized",
+			oldInitializers: nil,
+			newInitializers: &metav1.Initializers{Pending: []metav1.Initializer{{Name: "init.k8s.io"}}},
+			err:             "field is immutable once initialization has completed",
+		},
+	}
+
+	plugin := initializer{
+		config:     nil,
+		authorizer: &fakeAuthorizer{true},
+	}
+	for _, tc := range tests {
+		oldObj := &v1.Pod{}
+		oldObj.Initializers = tc.oldInitializers
+		newObj := &v1.Pod{}
+		newObj.Initializers = tc.newInitializers
+		a := admission.NewAttributesRecord(newObj, oldObj, schema.GroupVersionKind{}, "", "foo", schema.GroupVersionResource{}, "", admission.Update, nil)
+		err := plugin.Admit(a)
+		switch {
+		case tc.err == "" && err != nil:
+			t.Errorf("%q: unexpected error: %v", tc.name, err)
+		case tc.err != "" && err == nil:
+			t.Errorf("%q: unexpected no error, expected %s", tc.name, tc.err)
+		case tc.err != "" && err != nil && !strings.Contains(err.Error(), tc.err):
+			t.Errorf("%q: expected %s, got %v", tc.name, tc.err, err)
+		}
+	}
+
+}

staging/src/k8s.io/apimachinery/pkg/api/validation/objectmeta.go

@@ -200,9 +200,6 @@ func ValidateInitializers(initializers *metav1.Initializers, fldPath *field.Path
 		}
 	}
 	allErrs = append(allErrs, validateInitializersResult(initializers.Result, fldPath.Child("result"))...)
-	if len(initializers.Pending) == 0 && initializers.Result == nil {
-		allErrs = append(allErrs, field.Invalid(fldPath.Child("pending"), nil, "must be non-empty when result is not set"))
-	}
 	return allErrs
 }
 

staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/generated.proto

@@ -250,6 +250,8 @@ message Initializers {
   // When the last pending initializer is removed, and no failing result is set, the initializers
   // struct will be set to nil and the object is considered as initialized and visible to all
   // clients.
+  // +patchMergeKey=name
+  // +patchStrategy=merge
   repeated Initializer pending = 1;
 
   // If result is set with the Failure field, the object will be persisted to storage and then deleted,

staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/types.go

@@ -248,7 +248,9 @@ type Initializers struct {
 	// When the last pending initializer is removed, and no failing result is set, the initializers
 	// struct will be set to nil and the object is considered as initialized and visible to all
 	// clients.
-	Pending []Initializer `json:"pending" protobuf:"bytes,1,rep,name=pending"`
+	// +patchMergeKey=name
+	// +patchStrategy=merge
+	Pending []Initializer `json:"pending" protobuf:"bytes,1,rep,name=pending" patchStrategy:"merge" patchMergeKey:"name"`
 	// If result is set with the Failure field, the object will be persisted to storage and then deleted,
 	// ensuring that other clients can observe the deletion.
 	Result *Status `json:"result,omitempty" protobuf:"bytes,2,opt,name=result"`

test/e2e/apimachinery/initializers.go

@@ -262,6 +262,29 @@ var _ = SIGDescribe("Initializers", func() {
 		Expect(err).NotTo(HaveOccurred())
 		Expect(len(pods.Items)).Should(Equal(1))
 	})
+
+	It("will be set to nil if a patch removes the last pending initializer", func() {
+		ns := f.Namespace.Name
+		c := f.ClientSet
+
+		podName := "to-be-patch-initialized-pod"
+		framework.Logf("Creating pod %s", podName)
+
+		// TODO: lower the timeout so that the server responds faster.
+		_, err := c.CoreV1().Pods(ns).Create(newUninitializedPod(podName))
+		if err != nil && !errors.IsTimeout(err) {
+			framework.Failf("expect err to be timeout error, got %v", err)
+		}
+		uninitializedPod, err := c.CoreV1().Pods(ns).Get(podName, metav1.GetOptions{})
+		Expect(err).NotTo(HaveOccurred())
+		Expect(uninitializedPod.Initializers).NotTo(BeNil())
+		Expect(len(uninitializedPod.Initializers.Pending)).Should(Equal(1))
+
+		patch := fmt.Sprintf(`{"metadata":{"initializers":{"pending":[{"$patch":"delete","name":"%s"}]}}}`, uninitializedPod.Initializers.Pending[0].Name)
+		patchedPod, err := c.CoreV1().Pods(ns).Patch(uninitializedPod.Name, types.StrategicMergePatchType, []byte(patch))
+		Expect(err).NotTo(HaveOccurred())
+		Expect(patchedPod.Initializers).To(BeNil())
+	})
 })
 
 func newUninitializedPod(podName string) *v1.Pod {