Merge pull request #43034 from enisoc/statefulset-patch

Automatic merge from submit-queue (batch tested with PRs 43034, 43066)

Allow StatefulSet controller to PATCH Pods.

**What this PR does / why we need it**:

StatefulSet now needs the PATCH permission on Pods since it calls into ControllerRefManager to adopt and release. This adds the permission and the missing e2e test that should have caught this.

**Which issue this PR fixes**:

**Special notes for your reviewer**:

This is based on #42925.

**Release note**:
```release-note
```
cc @kubernetes/sig-apps-pr-reviews

plugin/pkg/auth/authorizer/rbac/bootstrappolicy/controller_policy.go

@@ -260,7 +260,7 @@ func init() {
 			rbac.NewRule("list", "watch").Groups(legacyGroup).Resources("pods").RuleOrDie(),
 			rbac.NewRule("get", "list", "watch").Groups(appsGroup).Resources("statefulsets").RuleOrDie(),
 			rbac.NewRule("update").Groups(appsGroup).Resources("statefulsets/status").RuleOrDie(),
-			rbac.NewRule("get", "create", "delete", "update").Groups(legacyGroup).Resources("pods").RuleOrDie(),
+			rbac.NewRule("get", "create", "delete", "update", "patch").Groups(legacyGroup).Resources("pods").RuleOrDie(),
 			rbac.NewRule("get", "create").Groups(legacyGroup).Resources("persistentvolumeclaims").RuleOrDie(),
 			eventsRule(),
 		},

plugin/pkg/auth/authorizer/rbac/bootstrappolicy/testdata/controller-roles.yaml

@@ -946,6 +946,7 @@ items:
     - create
     - delete
     - get
+    - patch
     - update
   - apiGroups:
     - ""

test/e2e/statefulset.go

@@ -33,6 +33,7 @@ import (
 	"k8s.io/kubernetes/pkg/api/v1"
 	apps "k8s.io/kubernetes/pkg/apis/apps/v1beta1"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
+	"k8s.io/kubernetes/pkg/controller"
 	"k8s.io/kubernetes/test/e2e/framework"
 )
 
@@ -127,6 +128,90 @@ var _ = framework.KubeDescribe("StatefulSet", func() {
 			framework.ExpectNoError(sst.ExecInStatefulPods(ss, cmd))
 		})
 
+		It("should adopt matching orphans and release non-matching pods", func() {
+			By("Creating statefulset " + ssName + " in namespace " + ns)
+			*(ss.Spec.Replicas) = 1
+			framework.SetStatefulSetInitializedAnnotation(ss, "false")
+
+			// Replace ss with the one returned from Create() so it has the UID.
+			// Save Kind since it won't be populated in the returned ss.
+			kind := ss.Kind
+			ss, err := c.Apps().StatefulSets(ns).Create(ss)
+			Expect(err).NotTo(HaveOccurred())
+			ss.Kind = kind
+
+			sst := framework.NewStatefulSetTester(c)
+
+			By("Saturating stateful set " + ss.Name)
+			sst.Saturate(ss)
+			pods := sst.GetPodList(ss)
+			Expect(pods.Items).To(HaveLen(int(*ss.Spec.Replicas)))
+
+			By("Checking that stateful set pods are created with ControllerRef")
+			pod := pods.Items[0]
+			controllerRef := controller.GetControllerOf(&pod)
+			Expect(controllerRef).ToNot(BeNil())
+			Expect(controllerRef.Kind).To(Equal(ss.Kind))
+			Expect(controllerRef.Name).To(Equal(ss.Name))
+			Expect(controllerRef.UID).To(Equal(ss.UID))
+
+			By("Orphaning one of the stateful set's pods")
+			f.PodClient().Update(pod.Name, func(pod *v1.Pod) {
+				pod.OwnerReferences = nil
+			})
+
+			By("Checking that the stateful set readopts the pod")
+			Expect(framework.WaitForPodCondition(c, pod.Namespace, pod.Name, "adopted", framework.StatefulSetTimeout,
+				func(pod *v1.Pod) (bool, error) {
+					controllerRef := controller.GetControllerOf(pod)
+					if controllerRef == nil {
+						return false, nil
+					}
+					if controllerRef.Kind != ss.Kind || controllerRef.Name != ss.Name || controllerRef.UID != ss.UID {
+						return false, fmt.Errorf("pod has wrong controllerRef: %v", controllerRef)
+					}
+					return true, nil
+				},
+			)).To(Succeed(), "wait for pod %q to be readopted", pod.Name)
+
+			By("Removing the labels from one of the stateful set's pods")
+			prevLabels := pod.Labels
+			f.PodClient().Update(pod.Name, func(pod *v1.Pod) {
+				pod.Labels = nil
+			})
+
+			By("Checking that the stateful set releases the pod")
+			Expect(framework.WaitForPodCondition(c, pod.Namespace, pod.Name, "released", framework.StatefulSetTimeout,
+				func(pod *v1.Pod) (bool, error) {
+					controllerRef := controller.GetControllerOf(pod)
+					if controllerRef != nil {
+						return false, nil
+					}
+					return true, nil
+				},
+			)).To(Succeed(), "wait for pod %q to be released", pod.Name)
+
+			// If we don't do this, the test leaks the Pod and PVC.
+			By("Readding labels to the stateful set's pod")
+			f.PodClient().Update(pod.Name, func(pod *v1.Pod) {
+				pod.Labels = prevLabels
+			})
+
+			By("Checking that the stateful set readopts the pod")
+			Expect(framework.WaitForPodCondition(c, pod.Namespace, pod.Name, "adopted", framework.StatefulSetTimeout,
+				func(pod *v1.Pod) (bool, error) {
+					controllerRef := controller.GetControllerOf(pod)
+					if controllerRef == nil {
+						return false, nil
+					}
+					if controllerRef.Kind != ss.Kind || controllerRef.Name != ss.Name || controllerRef.UID != ss.UID {
+						return false, fmt.Errorf("pod has wrong controllerRef: %v", controllerRef)
+					}
+					return true, nil
+				},
+			)).To(Succeed(), "wait for pod %q to be readopted", pod.Name)
+		})
+
 		It("should not deadlock when a pod's predecessor fails", func() {
 			By("Creating statefulset " + ssName + " in namespace " + ns)
 			*(ss.Spec.Replicas) = 2