Merge pull request #41361 from enisoc/controller-ref-manager

pkg/controller/BUILD

@@ -40,6 +40,7 @@ go_library(
         "//vendor:k8s.io/apimachinery/pkg/runtime",
         "//vendor:k8s.io/apimachinery/pkg/runtime/schema",
         "//vendor:k8s.io/apimachinery/pkg/types",
+        "//vendor:k8s.io/apimachinery/pkg/util/errors",
         "//vendor:k8s.io/apimachinery/pkg/util/sets",
         "//vendor:k8s.io/apimachinery/pkg/util/strategicpatch",
         "//vendor:k8s.io/apimachinery/pkg/util/wait",

pkg/controller/deployment/deployment_controller.go

@@ -443,11 +443,11 @@ func (dc *DeploymentController) handleErr(err error, key interface{}) {
 	dc.queue.Forget(key)
 }
 
-// classifyReplicaSets uses NewReplicaSetControllerRefManager to classify ReplicaSets
+// claimReplicaSets uses NewReplicaSetControllerRefManager to classify ReplicaSets
 // and adopts them if their labels match the Deployment but are missing the reference.
 // It also removes the controllerRef for ReplicaSets, whose labels no longer matches
 // the deployment.
-func (dc *DeploymentController) classifyReplicaSets(deployment *extensions.Deployment) error {
+func (dc *DeploymentController) claimReplicaSets(deployment *extensions.Deployment) error {
 	rsList, err := dc.rsLister.ReplicaSets(deployment.Namespace).List(labels.Everything())
 	if err != nil {
 		return err
@@ -457,32 +457,9 @@ func (dc *DeploymentController) classifyReplicaSets(deployment *extensions.Deplo
 	if err != nil {
 		return fmt.Errorf("deployment %s/%s has invalid label selector: %v", deployment.Namespace, deployment.Name, err)
 	}
-	cm := controller.NewReplicaSetControllerRefManager(dc.rsControl, deployment.ObjectMeta, deploymentSelector, getDeploymentKind())
-	matchesAndControlled, matchesNeedsController, controlledDoesNotMatch := cm.Classify(rsList)
-	// Adopt replica sets only if this deployment is not going to be deleted.
-	if deployment.DeletionTimestamp == nil {
-		for _, replicaSet := range matchesNeedsController {
-			err := cm.AdoptReplicaSet(replicaSet)
-			// continue to next RS if adoption fails.
-			if err != nil {
-				// If the RS no longer exists, don't even log the error.
-				if !errors.IsNotFound(err) {
-					utilruntime.HandleError(err)
-				}
-			} else {
-				matchesAndControlled = append(matchesAndControlled, replicaSet)
-			}
-		}
-	}
-	// remove the controllerRef for the RS that no longer have matching labels
-	var errlist []error
-	for _, replicaSet := range controlledDoesNotMatch {
-		err := cm.ReleaseReplicaSet(replicaSet)
-		if err != nil {
-			errlist = append(errlist, err)
-		}
-	}
-	return utilerrors.NewAggregate(errlist)
+	cm := controller.NewReplicaSetControllerRefManager(dc.rsControl, deployment, deploymentSelector, getDeploymentKind())
+	_, err = cm.ClaimReplicaSets(rsList)
+	return err
 }
 
 // syncDeployment will sync the deployment with the given key.
@@ -565,7 +542,7 @@ func (dc *DeploymentController) syncDeployment(key string) error {
 		dc.cleanupDeployment(oldRSs, d)
 	}
 
-	err = dc.classifyReplicaSets(d)
+	err = dc.claimReplicaSets(d)
 	if err != nil {
 		return err
 	}

pkg/controller/replicaset/BUILD

@@ -33,7 +33,6 @@ go_library(
         "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
         "//vendor:k8s.io/apimachinery/pkg/labels",
         "//vendor:k8s.io/apimachinery/pkg/runtime/schema",
-        "//vendor:k8s.io/apimachinery/pkg/util/errors",
         "//vendor:k8s.io/apimachinery/pkg/util/runtime",
         "//vendor:k8s.io/apimachinery/pkg/util/wait",
         "//vendor:k8s.io/client-go/kubernetes/typed/core/v1",

pkg/controller/replicaset/replica_set.go

@@ -30,7 +30,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	v1core "k8s.io/client-go/kubernetes/typed/core/v1"
@@ -603,38 +602,13 @@ func (rsc *ReplicaSetController) syncReplicaSet(key string) error {
 		if err != nil {
 			return err
 		}
-		cm := controller.NewPodControllerRefManager(rsc.podControl, rs.ObjectMeta, selector, getRSKind())
-		matchesAndControlled, matchesNeedsController, controlledDoesNotMatch := cm.Classify(pods)
-		// Adopt pods only if this replica set is not going to be deleted.
-		if rs.DeletionTimestamp == nil {
-			for _, pod := range matchesNeedsController {
-				err := cm.AdoptPod(pod)
-				// continue to next pod if adoption fails.
-				if err != nil {
-					// If the pod no longer exists, don't even log the error.
-					if !errors.IsNotFound(err) {
-						utilruntime.HandleError(err)
-					}
-				} else {
-					matchesAndControlled = append(matchesAndControlled, pod)
-				}
-			}
-		}
-		filteredPods = matchesAndControlled
-		// remove the controllerRef for the pods that no longer have matching labels
-		var errlist []error
-		for _, pod := range controlledDoesNotMatch {
-			err := cm.ReleasePod(pod)
-			if err != nil {
-				errlist = append(errlist, err)
-			}
-		}
-		if len(errlist) != 0 {
-			aggregate := utilerrors.NewAggregate(errlist)
-			// push the RS into work queue again. We need to try to free the
-			// pods again otherwise they will stuck with the stale
-			// controllerRef.
-			return aggregate
+		cm := controller.NewPodControllerRefManager(rsc.podControl, rs, selector, getRSKind())
+		filteredPods, err = cm.ClaimPods(pods)
+		if err != nil {
+			// Something went wrong with adoption or release.
+			// Requeue and try again so we don't leave orphans sitting around.
+			rsc.queue.Add(key)
+			return err
 		}
 	} else {
 		pods, err := rsc.podLister.Pods(rs.Namespace).List(selector)

pkg/controller/replicaset/replica_set_test.go

@@ -1224,14 +1224,20 @@ func TestPatchPodFails(t *testing.T) {
 	informers.Core().V1().Pods().Informer().GetIndexer().Add(newPod("pod1", rs, v1.PodRunning, nil))
 	informers.Core().V1().Pods().Informer().GetIndexer().Add(newPod("pod2", rs, v1.PodRunning, nil))
 	// let both patches fail. The rs controller will assume it fails to take
-	// control of the pods and create new ones.
+	// control of the pods and requeue to try again.
 	fakePodControl.Err = fmt.Errorf("Fake Error")
-	err := manager.syncReplicaSet(getKey(rs, t))
-	if err == nil || err.Error() != "Fake Error" {
+	rsKey := getKey(rs, t)
+	err := manager.syncReplicaSet(rsKey)
+	if err == nil || !strings.Contains(err.Error(), "Fake Error") {
 		t.Errorf("expected Fake Error, got %+v", err)
 	}
-	// 2 patches to take control of pod1 and pod2 (both fail), 2 creates.
-	validateSyncReplicaSet(t, fakePodControl, 2, 0, 2)
+	// 2 patches to take control of pod1 and pod2 (both fail).
+	validateSyncReplicaSet(t, fakePodControl, 0, 0, 2)
+	// RS should requeue itself.
+	queueRS, _ := manager.queue.Get()
+	if queueRS != rsKey {
+		t.Fatalf("Expected to find key %v in queue, found %v", rsKey, queueRS)
+	}
 }
 
 func TestPatchExtraPodsThenDelete(t *testing.T) {

pkg/controller/replication/BUILD

@@ -30,7 +30,6 @@ go_library(
         "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
         "//vendor:k8s.io/apimachinery/pkg/labels",
         "//vendor:k8s.io/apimachinery/pkg/runtime/schema",
-        "//vendor:k8s.io/apimachinery/pkg/util/errors",
         "//vendor:k8s.io/apimachinery/pkg/util/runtime",
         "//vendor:k8s.io/apimachinery/pkg/util/wait",
         "//vendor:k8s.io/apiserver/pkg/util/trace",

pkg/controller/replication/replication_controller.go

@@ -30,7 +30,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	utiltrace "k8s.io/apiserver/pkg/util/trace"
@@ -620,39 +619,13 @@ func (rm *ReplicationManager) syncReplicationController(key string) error {
 			rm.queue.Add(key)
 			return err
 		}
-		cm := controller.NewPodControllerRefManager(rm.podControl, rc.ObjectMeta, labels.Set(rc.Spec.Selector).AsSelectorPreValidated(), getRCKind())
-		matchesAndControlled, matchesNeedsController, controlledDoesNotMatch := cm.Classify(pods)
-		// Adopt pods only if this replication controller is not going to be deleted.
-		if rc.DeletionTimestamp == nil {
-			for _, pod := range matchesNeedsController {
-				err := cm.AdoptPod(pod)
-				// continue to next pod if adoption fails.
-				if err != nil {
-					// If the pod no longer exists, don't even log the error.
-					if !errors.IsNotFound(err) {
-						utilruntime.HandleError(err)
-					}
-				} else {
-					matchesAndControlled = append(matchesAndControlled, pod)
-				}
-			}
-		}
-		filteredPods = matchesAndControlled
-		// remove the controllerRef for the pods that no longer have matching labels
-		var errlist []error
-		for _, pod := range controlledDoesNotMatch {
-			err := cm.ReleasePod(pod)
-			if err != nil {
-				errlist = append(errlist, err)
-			}
-		}
-		if len(errlist) != 0 {
-			aggregate := utilerrors.NewAggregate(errlist)
-			// push the RC into work queue again. We need to try to free the
-			// pods again otherwise they will stuck with the stale
-			// controllerRef.
+		cm := controller.NewPodControllerRefManager(rm.podControl, rc, labels.Set(rc.Spec.Selector).AsSelectorPreValidated(), getRCKind())
+		filteredPods, err = cm.ClaimPods(pods)
+		if err != nil {
+			// Something went wrong with adoption or release.
+			// Requeue and try again so we don't leave orphans sitting around.
 			rm.queue.Add(key)
-			return aggregate
+			return err
 		}
 	} else {
 		pods, err := rm.podLister.Pods(rc.Namespace).List(labels.Set(rc.Spec.Selector).AsSelectorPreValidated())

pkg/controller/replication/replication_controller_test.go

@@ -1190,14 +1190,20 @@ func TestPatchPodFails(t *testing.T) {
 	podInformer.Informer().GetIndexer().Add(newPod("pod1", rc, v1.PodRunning, nil))
 	podInformer.Informer().GetIndexer().Add(newPod("pod2", rc, v1.PodRunning, nil))
 	// let both patches fail. The rc manager will assume it fails to take
-	// control of the pods and create new ones.
+	// control of the pods and requeue to try again.
 	fakePodControl.Err = fmt.Errorf("Fake Error")
-	err := manager.syncReplicationController(getKey(rc, t))
-	if err == nil || err.Error() != "Fake Error" {
+	rcKey := getKey(rc, t)
+	err := manager.syncReplicationController(rcKey)
+	if err == nil || !strings.Contains(err.Error(), "Fake Error") {
 		t.Fatalf("expected Fake Error, got %v", err)
 	}
-	// 2 patches to take control of pod1 and pod2 (both fail), 2 creates.
-	validateSyncReplication(t, fakePodControl, 2, 0, 2)
+	// 2 patches to take control of pod1 and pod2 (both fail).
+	validateSyncReplication(t, fakePodControl, 0, 0, 2)
+	// RC should requeue itself.
+	queueRC, _ := manager.queue.Get()
+	if queueRC != rcKey {
+		t.Fatalf("Expected to find key %v in queue, found %v", rcKey, queueRC)
+	}
 }
 
 func TestPatchExtraPodsThenDelete(t *testing.T) {