[GarbageCollector] Let the RC manager set/remove ControllerRef

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

@@ -215,6 +215,7 @@ func StartControllers(s *options.CMServer, kubeClient *client.Client, kubeconfig
 		ResyncPeriod(s),
 		replicationcontroller.BurstReplicas,
 		int(s.LookupCacheSizeForRC),
+		s.EnableGarbageCollector,
 	).Run(int(s.ConcurrentRCSyncs), wait.NeverStop)
 	time.Sleep(wait.Jitter(s.ControllerStartInterval.Duration, ControllerStartJitter))
 

pkg/controller/controller_ref_manager.go

@@ -0,0 +1,144 @@
+/*
+Copyright 2016 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/golang/glog"
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/api/errors"
+	"k8s.io/kubernetes/pkg/api/unversioned"
+	"k8s.io/kubernetes/pkg/labels"
+)
+
+type PodControllerRefManager struct {
+	podControl         PodControlInterface
+	controllerObject   api.ObjectMeta
+	controllerSelector labels.Selector
+	controllerKind     unversioned.GroupVersionKind
+}
+
+// NewPodControllerRefManager returns a PodControllerRefManager that exposes
+// methods to manage the controllerRef of pods.
+func NewPodControllerRefManager(
+	podControl PodControlInterface,
+	controllerObject api.ObjectMeta,
+	controllerSelector labels.Selector,
+	controllerKind unversioned.GroupVersionKind,
+) *PodControllerRefManager {
+	return &PodControllerRefManager{podControl, controllerObject, controllerSelector, controllerKind}
+}
+
+// Classify first filters out inactive pods, then it classify the remaining pods
+// into three categories: 1. matchesAndControlled are the pods whose labels
+// match the selector of the RC, and have a controllerRef pointing to the
+// controller 2. matchesNeedsController are the pods whose labels match the RC,
+// but don't have a controllerRef. (Pods with matching labels but with a
+// controllerRef pointing to other object are ignored) 3. controlledDoesNotMatch
+// are the pods that have a controllerRef pointing to the controller, but their
+// labels no longer match the selector.
+func (m *PodControllerRefManager) Classify(pods []api.Pod) (
+	matchesAndControlled []*api.Pod,
+	matchesNeedsController []*api.Pod,
+	controlledDoesNotMatch []*api.Pod) {
+	for i := range pods {
+		pod := pods[i]
+		if !IsPodActive(pod) {
+			glog.V(4).Infof("Ignoring inactive pod %v/%v in state %v, deletion time %v",
+				pod.Namespace, pod.Name, pod.Status.Phase, pod.DeletionTimestamp)
+			continue
+		}
+		controllerRef := getControllerOf(pod.ObjectMeta)
+		if controllerRef != nil {
+			if controllerRef.UID == m.controllerObject.UID {
+				// already controlled
+				if m.controllerSelector.Matches(labels.Set(pod.Labels)) {
+					matchesAndControlled = append(matchesAndControlled, &pod)
+				} else {
+					controlledDoesNotMatch = append(controlledDoesNotMatch, &pod)
+				}
+			} else {
+				// ignoring the pod controlled by other controller
+				glog.V(4).Infof("Ignoring pod %v/%v, it's owned by [%s/%s, name: %s, uid: %s]",
+					pod.Namespace, pod.Name, controllerRef.APIVersion, controllerRef.Kind, controllerRef.Name, controllerRef.UID)
+				continue
+			}
+		} else {
+			if !m.controllerSelector.Matches(labels.Set(pod.Labels)) {
+				continue
+			}
+			matchesNeedsController = append(matchesNeedsController, &pod)
+		}
+	}
+	return matchesAndControlled, matchesNeedsController, controlledDoesNotMatch
+}
+
+// getControllerOf returns the controllerRef if controllee has a controller,
+// otherwise returns nil.
+func getControllerOf(controllee api.ObjectMeta) *api.OwnerReference {
+	for _, owner := range controllee.OwnerReferences {
+		// controlled by other controller
+		if owner.Controller != nil && *owner.Controller == true {
+			return &owner
+		}
+	}
+	return nil
+}
+
+// AdoptPod sends a patch to take control of the pod. It returns the error if
+// the patching fails.
+func (m *PodControllerRefManager) AdoptPod(pod *api.Pod) error {
+	// we should not adopt any pods if the controller is about to be deleted
+	if m.controllerObject.DeletionTimestamp != nil {
+		return fmt.Errorf("cancel the adopt attempt for pod %s because the controlller is being deleted",
+			strings.Join([]string{pod.Namespace, pod.Name, string(pod.UID)}, "_"))
+	}
+	addControllerPatch := fmt.Sprintf(
+		`{"metadata":{"ownerReferences":[{"apiVersion":"%s","kind":"%s","name":"%s","uid":"%s","controller":true}],"uid":"%s"}}`,
+		m.controllerKind.GroupVersion(), m.controllerKind.Kind,
+		m.controllerObject.Name, m.controllerObject.UID, pod.UID)
+	return m.podControl.PatchPod(pod.Namespace, pod.Name, []byte(addControllerPatch))
+}
+
+// ReleasePod sends a patch to free the pod from the control of the controller.
+// It returns the error if the patching fails. 404 and 422 errors are ignored.
+func (m *PodControllerRefManager) ReleasePod(pod *api.Pod) error {
+	glog.V(2).Infof("patching pod %s_%s to remove its controllerRef to %s/%s:%s",
+		pod.Namespace, pod.Name, m.controllerKind.GroupVersion(), m.controllerKind.Kind, m.controllerObject.Name)
+	deleteOwnerRefPatch := fmt.Sprintf(`{"metadata":{"ownerReferences":[{"$patch":"delete","uid":"%s"}],"uid":"%s"}}`, m.controllerObject.UID, pod.UID)
+	err := m.podControl.PatchPod(pod.Namespace, pod.Name, []byte(deleteOwnerRefPatch))
+	if err != nil {
+		if errors.IsNotFound(err) {
+			// If the pod no longer exists, ignore it.
+			return nil
+		}
+		if errors.IsInvalid(err) {
+			// Invalid error will be returned in two cases: 1. the pod
+			// has no owner reference, 2. the uid of the pod doesn't
+			// match, which means the pod is deleted and then recreated.
+			// In both cases, the error can be ignored.
+
+			// TODO: If the pod has owner references, but none of them
+			// has the owner.UID, server will silently ignore the patch.
+			// Investigate why.
+			return nil
+		}
+	}
+	return err
+}

pkg/controller/controller_utils.go

@@ -349,8 +349,12 @@ type PodControlInterface interface {
 	CreatePods(namespace string, template *api.PodTemplateSpec, object runtime.Object) error
 	// CreatePodsOnNode creates a new pod accorting to the spec on the specified node.
 	CreatePodsOnNode(nodeName, namespace string, template *api.PodTemplateSpec, object runtime.Object) error
+	// CreatePodsWithControllerRef creates new pods according to the spec, and sets object as the pod's controller.
+	CreatePodsWithControllerRef(namespace string, template *api.PodTemplateSpec, object runtime.Object, controllerRef *api.OwnerReference) error
 	// DeletePod deletes the pod identified by podID.
 	DeletePod(namespace string, podID string, object runtime.Object) error
+	// PatchPod patches the pod.
+	PatchPod(namespace, name string, data []byte) error
 }
 
 // RealPodControl is the default implementation of PodControlInterface.
@@ -404,14 +408,35 @@ func getPodsPrefix(controllerName string) string {
 }
 
 func (r RealPodControl) CreatePods(namespace string, template *api.PodTemplateSpec, object runtime.Object) error {
-	return r.createPods("", namespace, template, object)
+	return r.createPods("", namespace, template, object, nil)
+}
+
+func (r RealPodControl) CreatePodsWithControllerRef(namespace string, template *api.PodTemplateSpec, controllerObject runtime.Object, controllerRef *api.OwnerReference) error {
+	if controllerRef == nil {
+		return fmt.Errorf("controllerRef is nil")
+	}
+	if len(controllerRef.APIVersion) == 0 {
+		return fmt.Errorf("controllerRef has empty APIVersion")
+	}
+	if len(controllerRef.Kind) == 0 {
+		return fmt.Errorf("controllerRef has empty Kind")
+	}
+	if controllerRef.Controller == nil || *controllerRef.Controller != true {
+		return fmt.Errorf("controllerRef.Controller is not set")
+	}
+	return r.createPods("", namespace, template, controllerObject, controllerRef)
 }
 
 func (r RealPodControl) CreatePodsOnNode(nodeName, namespace string, template *api.PodTemplateSpec, object runtime.Object) error {
-	return r.createPods(nodeName, namespace, template, object)
+	return r.createPods(nodeName, namespace, template, object, nil)
+}
+
+func (r RealPodControl) PatchPod(namespace, name string, data []byte) error {
+	_, err := r.KubeClient.Core().Pods(namespace).Patch(name, api.StrategicMergePatchType, data)
+	return err
 }
 
-func GetPodFromTemplate(template *api.PodTemplateSpec, parentObject runtime.Object) (*api.Pod, error) {
+func GetPodFromTemplate(template *api.PodTemplateSpec, parentObject runtime.Object, controllerRef *api.OwnerReference) (*api.Pod, error) {
 	desiredLabels := getPodsLabelSet(template)
 	desiredAnnotations, err := getPodsAnnotationSet(template, parentObject)
 	if err != nil {
@@ -430,14 +455,17 @@ func GetPodFromTemplate(template *api.PodTemplateSpec, parentObject runtime.Obje
 			GenerateName: prefix,
 		},
 	}
+	if controllerRef != nil {
+		pod.OwnerReferences = append(pod.OwnerReferences, *controllerRef)
+	}
 	if err := api.Scheme.Convert(&template.Spec, &pod.Spec); err != nil {
 		return nil, fmt.Errorf("unable to convert pod template: %v", err)
 	}
 	return pod, nil
 }
 
-func (r RealPodControl) createPods(nodeName, namespace string, template *api.PodTemplateSpec, object runtime.Object) error {
-	pod, err := GetPodFromTemplate(template, object)
+func (r RealPodControl) createPods(nodeName, namespace string, template *api.PodTemplateSpec, object runtime.Object, controllerRef *api.OwnerReference) error {
+	pod, err := GetPodFromTemplate(template, object, controllerRef)
 	if err != nil {
 		return err
 	}
@@ -479,40 +507,63 @@ func (r RealPodControl) DeletePod(namespace string, podID string, object runtime
 
 type FakePodControl struct {
 	sync.Mutex
-	Templates     []api.PodTemplateSpec
-	DeletePodName []string
-	Err           error
+	Templates      []api.PodTemplateSpec
+	ControllerRefs []api.OwnerReference
+	DeletePodName  []string
+	Patches        [][]byte
+	Err            error
 }
 
 var _ PodControlInterface = &FakePodControl{}
 
+func (f *FakePodControl) PatchPod(namespace, name string, data []byte) error {
+	f.Lock()
+	defer f.Unlock()
+	f.Patches = append(f.Patches, data)
+	if f.Err != nil {
+		return f.Err
+	}
+	return nil
+}
+
 func (f *FakePodControl) CreatePods(namespace string, spec *api.PodTemplateSpec, object runtime.Object) error {
 	f.Lock()
 	defer f.Unlock()
+	f.Templates = append(f.Templates, *spec)
 	if f.Err != nil {
 		return f.Err
 	}
+	return nil
+}
+
+func (f *FakePodControl) CreatePodsWithControllerRef(namespace string, spec *api.PodTemplateSpec, object runtime.Object, controllerRef *api.OwnerReference) error {
+	f.Lock()
+	defer f.Unlock()
 	f.Templates = append(f.Templates, *spec)
+	f.ControllerRefs = append(f.ControllerRefs, *controllerRef)
+	if f.Err != nil {
+		return f.Err
+	}
 	return nil
 }
 
 func (f *FakePodControl) CreatePodsOnNode(nodeName, namespace string, template *api.PodTemplateSpec, object runtime.Object) error {
 	f.Lock()
 	defer f.Unlock()
+	f.Templates = append(f.Templates, *template)
 	if f.Err != nil {
 		return f.Err
 	}
-	f.Templates = append(f.Templates, *template)
 	return nil
 }
 
 func (f *FakePodControl) DeletePod(namespace string, podID string, object runtime.Object) error {
 	f.Lock()
 	defer f.Unlock()
+	f.DeletePodName = append(f.DeletePodName, podID)
 	if f.Err != nil {
 		return f.Err
 	}
-	f.DeletePodName = append(f.DeletePodName, podID)
 	return nil
 }
 
@@ -521,6 +572,8 @@ func (f *FakePodControl) Clear() {
 	defer f.Unlock()
 	f.DeletePodName = []string{}
 	f.Templates = []api.PodTemplateSpec{}
+	f.ControllerRefs = []api.OwnerReference{}
+	f.Patches = [][]byte{}
 }
 
 // ByLogging allows custom sorting of pods so the best one can be picked for getting its logs.

pkg/controller/controller_utils_test.go

@@ -253,7 +253,9 @@ func TestCreatePods(t *testing.T) {
 	controllerSpec := newReplicationController(1)
 
 	// Make sure createReplica sends a POST to the apiserver with a pod from the controllers pod template
-	podControl.CreatePods(ns, controllerSpec.Spec.Template, controllerSpec)
+	if err := podControl.CreatePods(ns, controllerSpec.Spec.Template, controllerSpec); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
 
 	expectedPod := api.Pod{
 		ObjectMeta: api.ObjectMeta{
@@ -265,7 +267,7 @@ func TestCreatePods(t *testing.T) {
 	fakeHandler.ValidateRequest(t, testapi.Default.ResourcePath("pods", api.NamespaceDefault, ""), "POST", nil)
 	actualPod, err := runtime.Decode(testapi.Default.Codec(), []byte(fakeHandler.RequestBody))
 	if err != nil {
-		t.Errorf("Unexpected error: %#v", err)
+		t.Fatalf("Unexpected error: %v", err)
 	}
 	if !api.Semantic.DeepDerivative(&expectedPod, actualPod) {
 		t.Logf("Body: %s", fakeHandler.RequestBody)

pkg/controller/job/jobcontroller_test.go

@@ -164,7 +164,7 @@ func TestControllerSyncJob(t *testing.T) {
 		"too few active pods, with controller error": {
 			2, 5, false,
 			fmt.Errorf("Fake error"), 0, 1, 1, 0,
-			0, 0, 1, 1, 0, false,
+			1, 0, 1, 1, 0, false,
 		},
 		"too many active pods": {
 			2, 5, false,
@@ -174,7 +174,7 @@ func TestControllerSyncJob(t *testing.T) {
 		"too many active pods, with controller error": {
 			2, 5, false,
 			fmt.Errorf("Fake error"), 0, 3, 0, 0,
-			0, 0, 3, 0, 0, false,
+			0, 1, 3, 0, 0, false,
 		},
 		"failed pod": {
 			2, 5, false,

pkg/controller/petset/iterator.go

@@ -28,7 +28,7 @@ import (
 
 // newPCB generates a new PCB using the id string as a unique qualifier
 func newPCB(id string, ps *apps.PetSet) (*pcb, error) {
-	petPod, err := controller.GetPodFromTemplate(&ps.Spec.Template, ps)
+	petPod, err := controller.GetPodFromTemplate(&ps.Spec.Template, ps, nil)
 	if err != nil {
 		return nil, err
 	}

pkg/controller/replicaset/replica_set_test.go

@@ -352,9 +352,10 @@ func TestSyncReplicaSetDormancy(t *testing.T) {
 	fakePodControl.Err = fmt.Errorf("Fake Error")
 
 	manager.syncReplicaSet(getKey(rsSpec, t))
-	validateSyncReplicaSet(t, &fakePodControl, 0, 0)
+	validateSyncReplicaSet(t, &fakePodControl, 1, 0)
 
 	// This replica should not need a Lowering of expectations, since the previous create failed
+	fakePodControl.Clear()
 	fakePodControl.Err = nil
 	manager.syncReplicaSet(getKey(rsSpec, t))
 	validateSyncReplicaSet(t, &fakePodControl, 1, 0)