Skip to content

K
k3s
Commit 70bd5fdf authored by Anthony Yeh's avatar Anthony Yeh
Browse files
Options

Refactor ControllerRefManager

To prepare for implementing ControllerRef across all controllers, this pushes the common adopt/orphan logic into ControllerRefManager so each controller doesn't have to duplicate it. This also shares the adopt/orphan logic between Pods and ReplicaSets, so it lives in only one place.
parent 4ac2749a
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 42 additions and 107 deletions
pkg/controller/BUILD
View file @ 70bd5fdf
......@@ -38,6 +38,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
View file @ 70bd5fdf
......@@ -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
View file @ 70bd5fdf
......@@ -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
View file @ 70bd5fdf
......@@ -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
View file @ 70bd5fdf
......@@ -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
View file @ 70bd5fdf
......@@ -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
View file @ 70bd5fdf
......@@ -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
View file @ 70bd5fdf
......@@ -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) {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment