Skip to content

K
k3s
Commit 47320fd3 authored by Kubernetes Submit Queue's avatar Kubernetes Submit Queue Committed by GitHub
Browse files
Options

Merge pull request #42938 from enisoc/orphan-race

Automatic merge from submit-queue GC: Fix re-adoption race when orphaning dependents. **What this PR does / why we need it**: The GC expects that once it sees a controller with a non-nil DeletionTimestamp, that controller will not attempt any adoption. There was a known race condition that could cause a controller to re-adopt something orphaned by the GC, because the controller is using a cached value of its own spec from before DeletionTimestamp was set. This fixes that race by doing an uncached quorum read of the controller spec just before the first adoption attempt. It's important that this read occurs after listing potential orphans. Note that this uncached read is skipped if no adoptions are attempted (i.e. at steady state). **Which issue this PR fixes**: Fixes #42639 **Special notes for your reviewer**: **Release note**: ```release-note ``` cc @kubernetes/sig-apps-pr-reviews
parents f8803403 b4b8fdbc
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 376 additions and 176 deletions
pkg/controller/controller_ref_manager.go
View file @ 47320fd3
......@@ -18,7 +18,7 @@ package controller
import (
"fmt"
"strings"
"sync"
"github.com/golang/glog"
"k8s.io/apimachinery/pkg/api/errors"
......@@ -46,6 +46,19 @@ func GetControllerOf(controllee metav1.Object) *metav1.OwnerReference {
type baseControllerRefManager struct {
controller metav1.Object
selector labels.Selector
canAdoptErr error
canAdoptOnce sync.Once
canAdoptFunc func() error
}
func (m *baseControllerRefManager) canAdopt() error {
m.canAdoptOnce.Do(func() {
if m.canAdoptFunc != nil {
m.canAdoptErr = m.canAdoptFunc()
}
})
return m.canAdoptErr
}
// claimObject tries to take ownership of an object for this controller.
......@@ -122,16 +135,27 @@ type PodControllerRefManager struct {
// NewPodControllerRefManager returns a PodControllerRefManager that exposes
// methods to manage the controllerRef of pods.
//
// The canAdopt() function can be used to perform a potentially expensive check
// (such as a live GET from the API server) prior to the first adoption.
// It will only be called (at most once) if an adoption is actually attempted.
// If canAdopt() returns a non-nil error, all adoptions will fail.
//
// NOTE: Once canAdopt() is called, it will not be called again by the same
// PodControllerRefManager instance. Create a new instance if it makes
// sense to check canAdopt() again (e.g. in a different sync pass).
func NewPodControllerRefManager(
podControl PodControlInterface,
controller metav1.Object,
selector labels.Selector,
controllerKind schema.GroupVersionKind,
canAdopt func() error,
) *PodControllerRefManager {
return &PodControllerRefManager{
baseControllerRefManager: baseControllerRefManager{
controller: controller,
selector: selector,
controller: controller,
selector: selector,
canAdoptFunc: canAdopt,
},
controllerKind: controllerKind,
podControl: podControl,
......@@ -193,10 +217,8 @@ func (m *PodControllerRefManager) ClaimPods(pods []*v1.Pod, filters ...func(*v1.
// AdoptPod sends a patch to take control of the pod. It returns the error if
// the patching fails.
func (m *PodControllerRefManager) AdoptPod(pod *v1.Pod) error {
// we should not adopt any pods if the controller is about to be deleted
if m.controller.GetDeletionTimestamp() != nil {
return fmt.Errorf("cancel the adopt attempt for pod %s because the controller is being deleted",
strings.Join([]string{pod.Namespace, pod.Name, string(pod.UID)}, "_"))
if err := m.canAdopt(); err != nil {
return fmt.Errorf("can't adopt Pod %v/%v (%v): %v", pod.Namespace, pod.Name, pod.UID, err)
}
// Note that ValidateOwnerReferences() will reject this patch if another
// OwnerReference exists with controller=true.
......@@ -247,16 +269,27 @@ type ReplicaSetControllerRefManager struct {
// NewReplicaSetControllerRefManager returns a ReplicaSetControllerRefManager that exposes
// methods to manage the controllerRef of ReplicaSets.
//
// The canAdopt() function can be used to perform a potentially expensive check
// (such as a live GET from the API server) prior to the first adoption.
// It will only be called (at most once) if an adoption is actually attempted.
// If canAdopt() returns a non-nil error, all adoptions will fail.
//
// NOTE: Once canAdopt() is called, it will not be called again by the same
// ReplicaSetControllerRefManager instance. Create a new instance if it
// makes sense to check canAdopt() again (e.g. in a different sync pass).
func NewReplicaSetControllerRefManager(
rsControl RSControlInterface,
controller metav1.Object,
selector labels.Selector,
controllerKind schema.GroupVersionKind,
canAdopt func() error,
) *ReplicaSetControllerRefManager {
return &ReplicaSetControllerRefManager{
baseControllerRefManager: baseControllerRefManager{
controller: controller,
selector: selector,
controller: controller,
selector: selector,
canAdoptFunc: canAdopt,
},
controllerKind: controllerKind,
rsControl: rsControl,
......@@ -305,19 +338,17 @@ func (m *ReplicaSetControllerRefManager) ClaimReplicaSets(sets []*extensions.Rep
// AdoptReplicaSet sends a patch to take control of the ReplicaSet. It returns the error if
// the patching fails.
func (m *ReplicaSetControllerRefManager) AdoptReplicaSet(replicaSet *extensions.ReplicaSet) error {
// we should not adopt any ReplicaSets if the Deployment is about to be deleted
if m.controller.GetDeletionTimestamp() != nil {
return fmt.Errorf("cancel the adopt attempt for RS %s because the controller %v is being deleted",
strings.Join([]string{replicaSet.Namespace, replicaSet.Name, string(replicaSet.UID)}, "_"), m.controller.GetName())
func (m *ReplicaSetControllerRefManager) AdoptReplicaSet(rs *extensions.ReplicaSet) error {
if err := m.canAdopt(); err != nil {
return fmt.Errorf("can't adopt ReplicaSet %v/%v (%v): %v", rs.Namespace, rs.Name, rs.UID, err)
}
// Note that ValidateOwnerReferences() will reject this patch if another
// OwnerReference exists with controller=true.
addControllerPatch := fmt.Sprintf(
`{"metadata":{"ownerReferences":[{"apiVersion":"%s","kind":"%s","name":"%s","uid":"%s","controller":true,"blockOwnerDeletion":true}],"uid":"%s"}}`,
m.controllerKind.GroupVersion(), m.controllerKind.Kind,
m.controller.GetName(), m.controller.GetUID(), replicaSet.UID)
return m.rsControl.PatchReplicaSet(replicaSet.Namespace, replicaSet.Name, []byte(addControllerPatch))
m.controller.GetName(), m.controller.GetUID(), rs.UID)
return m.rsControl.PatchReplicaSet(rs.Namespace, rs.Name, []byte(addControllerPatch))
}
// ReleaseReplicaSet sends a patch to free the ReplicaSet from the control of the Deployment controller.
......@@ -342,3 +373,20 @@ func (m *ReplicaSetControllerRefManager) ReleaseReplicaSet(replicaSet *extension
}
return err
}
// RecheckDeletionTimestamp returns a canAdopt() function to recheck deletion.
//
// The canAdopt() function calls getObject() to fetch the latest value,
// and denies adoption attempts if that object has a non-nil DeletionTimestamp.
func RecheckDeletionTimestamp(getObject func() (metav1.Object, error)) func() error {
return func() error {
obj, err := getObject()
if err != nil {
return fmt.Errorf("can't recheck DeletionTimestamp: %v", err)
}
if obj.GetDeletionTimestamp() != nil {
return fmt.Errorf("%v/%v has just been deleted at %v", obj.GetNamespace(), obj.GetName(), obj.GetDeletionTimestamp())
}
return nil
}
}
pkg/controller/daemon/daemoncontroller.go
View file @ 47320fd3
......@@ -468,8 +468,20 @@ func (dsc *DaemonSetsController) getNodesToDaemonPods(ds *extensions.DaemonSet)
if err != nil {
return nil, err
}
// If any adoptions are attempted, we should first recheck for deletion with
// an uncached quorum read sometime after listing Pods (see #42639).
canAdoptFunc := controller.RecheckDeletionTimestamp(func() (metav1.Object, error) {
fresh, err := dsc.kubeClient.ExtensionsV1beta1().DaemonSets(ds.Namespace).Get(ds.Name, metav1.GetOptions{})
if err != nil {
return nil, err
}
if fresh.UID != ds.UID {
return nil, fmt.Errorf("original DaemonSet %v/%v is gone: got uid %v, wanted %v", ds.Namespace, ds.Name, fresh.UID, ds.UID)
}
return fresh, nil
})
// Use ControllerRefManager to adopt/orphan as needed.
cm := controller.NewPodControllerRefManager(dsc.podControl, ds, selector, controllerKind)
cm := controller.NewPodControllerRefManager(dsc.podControl, ds, selector, controllerKind, canAdoptFunc)
claimedPods, err := cm.ClaimPods(pods)
if err != nil {
return nil, err
......
pkg/controller/daemon/update_test.go
View file @ 47320fd3
......@@ -24,10 +24,10 @@ import (
)
func TestDaemonSetUpdatesPods(t *testing.T) {
manager, podControl, _ := newTestController()
ds := newDaemonSet("foo")
manager, podControl, _ := newTestController(ds)
maxUnavailable := 2
addNodes(manager.nodeStore, 0, 5, nil)
ds := newDaemonSet("foo")
manager.dsStore.Add(ds)
syncAndValidateDaemonSets(t, manager, ds, podControl, 5, 0)
markPodsReady(podControl.podStore)
......@@ -63,10 +63,10 @@ func TestDaemonSetUpdatesPods(t *testing.T) {
}
func TestDaemonSetUpdatesWhenNewPosIsNotReady(t *testing.T) {
manager, podControl, _ := newTestController()
ds := newDaemonSet("foo")
manager, podControl, _ := newTestController(ds)
maxUnavailable := 3
addNodes(manager.nodeStore, 0, 5, nil)
ds := newDaemonSet("foo")
manager.dsStore.Add(ds)
syncAndValidateDaemonSets(t, manager, ds, podControl, 5, 0)
markPodsReady(podControl.podStore)
......@@ -90,10 +90,10 @@ func TestDaemonSetUpdatesWhenNewPosIsNotReady(t *testing.T) {
}
func TestDaemonSetUpdatesAllOldPodsNotReady(t *testing.T) {
manager, podControl, _ := newTestController()
ds := newDaemonSet("foo")
manager, podControl, _ := newTestController(ds)
maxUnavailable := 3
addNodes(manager.nodeStore, 0, 5, nil)
ds := newDaemonSet("foo")
manager.dsStore.Add(ds)
syncAndValidateDaemonSets(t, manager, ds, podControl, 5, 0)
......@@ -116,10 +116,10 @@ func TestDaemonSetUpdatesAllOldPodsNotReady(t *testing.T) {
}
func TestDaemonSetUpdatesNoTemplateChanged(t *testing.T) {
manager, podControl, _ := newTestController()
ds := newDaemonSet("foo")
manager, podControl, _ := newTestController(ds)
maxUnavailable := 3
addNodes(manager.nodeStore, 0, 5, nil)
ds := newDaemonSet("foo")
manager.dsStore.Add(ds)
syncAndValidateDaemonSets(t, manager, ds, podControl, 5, 0)
......
pkg/controller/deployment/deployment_controller.go
View file @ 47320fd3
......@@ -501,7 +501,19 @@ func (dc *DeploymentController) getReplicaSetsForDeployment(d *extensions.Deploy
if err != nil {
return nil, fmt.Errorf("deployment %s/%s has invalid label selector: %v", d.Namespace, d.Name, err)
}
cm := controller.NewReplicaSetControllerRefManager(dc.rsControl, d, deploymentSelector, controllerKind)
// If any adoptions are attempted, we should first recheck for deletion with
// an uncached quorum read sometime after listing ReplicaSets (see #42639).
canAdoptFunc := controller.RecheckDeletionTimestamp(func() (metav1.Object, error) {
fresh, err := dc.client.ExtensionsV1beta1().Deployments(d.Namespace).Get(d.Name, metav1.GetOptions{})
if err != nil {
return nil, err
}
if fresh.UID != d.UID {
return nil, fmt.Errorf("original Deployment %v/%v is gone: got uid %v, wanted %v", d.Namespace, d.Name, fresh.UID, d.UID)
}
return fresh, nil
})
cm := controller.NewReplicaSetControllerRefManager(dc.rsControl, d, deploymentSelector, controllerKind, canAdoptFunc)
return cm.ClaimReplicaSets(rsList)
}
......
pkg/controller/deployment/deployment_controller_test.go
View file @ 47320fd3
......@@ -148,6 +148,11 @@ type fixture struct {
objects []runtime.Object
}
func (f *fixture) expectGetDeploymentAction(d *extensions.Deployment) {
action := core.NewGetAction(schema.GroupVersionResource{Resource: "deployments"}, d.Namespace, d.Name)
f.actions = append(f.actions, action)
}
func (f *fixture) expectUpdateDeploymentStatusAction(d *extensions.Deployment) {
action := core.NewUpdateAction(schema.GroupVersionResource{Resource: "deployments"}, d.Namespace, d)
action.Subresource = "status"
......@@ -190,15 +195,25 @@ func (f *fixture) newController() (*DeploymentController, informers.SharedInform
return c, informers
}
func (f *fixture) runExpectError(deploymentName string) {
f.run_(deploymentName, true)
}
func (f *fixture) run(deploymentName string) {
f.run_(deploymentName, false)
}
func (f *fixture) run_(deploymentName string, expectError bool) {
c, informers := f.newController()
stopCh := make(chan struct{})
defer close(stopCh)
informers.Start(stopCh)
err := c.syncDeployment(deploymentName)
if err != nil {
if !expectError && err != nil {
f.t.Errorf("error syncing deployment: %v", err)
} else if expectError && err == nil {
f.t.Error("expected error syncing deployment, got nil")
}
actions := filterInformerActions(f.client.Actions())
......@@ -293,6 +308,30 @@ func TestSyncDeploymentDontDoAnythingDuringDeletion(t *testing.T) {
f.run(getKey(d, t))
}
func TestSyncDeploymentDeletionRace(t *testing.T) {
f := newFixture(t)
d := newDeployment("foo", 1, nil, nil, nil, map[string]string{"foo": "bar"})
d2 := *d
// Lister (cache) says NOT deleted.
f.dLister = append(f.dLister, d)
// Bare client says it IS deleted. This should be presumed more up-to-date.
now := metav1.Now()
d2.DeletionTimestamp = &now
f.objects = append(f.objects, &d2)
// The recheck is only triggered if a matching orphan exists.
rs := newReplicaSet(d, "rs1", 1)
rs.OwnerReferences = nil
f.objects = append(f.objects, rs)
f.rsLister = append(f.rsLister, rs)
// Expect to only recheck DeletionTimestamp.
f.expectGetDeploymentAction(d)
// Sync should fail and requeue to let cache catch up.
f.runExpectError(getKey(d, t))
}
// issue: https://github.com/kubernetes/kubernetes/issues/23218
func TestDontSyncDeploymentsWithEmptyPodSelector(t *testing.T) {
f := newFixture(t)
......
pkg/controller/replicaset/BUILD
View file @ 47320fd3
......@@ -68,6 +68,7 @@ go_test(
"//vendor:k8s.io/client-go/testing",
"//vendor:k8s.io/client-go/tools/cache",
"//vendor:k8s.io/client-go/util/testing",
"//vendor:k8s.io/client-go/util/workqueue",
],
)
......
pkg/controller/replicaset/replica_set.go
View file @ 47320fd3
......@@ -564,7 +564,19 @@ func (rsc *ReplicaSetController) syncReplicaSet(key string) error {
filteredPods = append(filteredPods, pod)
}
}
cm := controller.NewPodControllerRefManager(rsc.podControl, rs, selector, controllerKind)
// If any adoptions are attempted, we should first recheck for deletion with
// an uncached quorum read sometime after listing Pods (see #42639).
canAdoptFunc := controller.RecheckDeletionTimestamp(func() (metav1.Object, error) {
fresh, err := rsc.kubeClient.ExtensionsV1beta1().ReplicaSets(rs.Namespace).Get(rs.Name, metav1.GetOptions{})
if err != nil {
return nil, err
}
if fresh.UID != rs.UID {
return nil, fmt.Errorf("original ReplicaSet %v/%v is gone: got uid %v, wanted %v", rs.Namespace, rs.Name, fresh.UID, rs.UID)
}
return fresh, nil
})
cm := controller.NewPodControllerRefManager(rsc.podControl, rs, selector, controllerKind, canAdoptFunc)
// NOTE: filteredPods are pointing to objects from cache - if you need to
// modify them, you need to copy it first.
filteredPods, err = cm.ClaimPods(filteredPods)
......
pkg/controller/replicaset/replica_set_test.go
View file @ 47320fd3
......@@ -19,6 +19,7 @@ limitations under the License.
package replicaset
import (
"errors"
"fmt"
"math/rand"
"net/http/httptest"
......@@ -39,6 +40,7 @@ import (
core "k8s.io/client-go/testing"
"k8s.io/client-go/tools/cache"
utiltesting "k8s.io/client-go/util/testing"
"k8s.io/client-go/util/workqueue"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/testapi"
"k8s.io/kubernetes/pkg/api/v1"
......@@ -87,7 +89,7 @@ func skipListerFunc(verb string, url url.URL) bool {
if verb != "GET" {
return false
}
if strings.HasSuffix(url.Path, "/pods") || strings.HasSuffix(url.Path, "/replicasets") {
if strings.HasSuffix(url.Path, "/pods") || strings.Contains(url.Path, "/replicasets") {
return true
}
return false
......@@ -247,7 +249,9 @@ func TestSyncReplicaSetDoesNothing(t *testing.T) {
}
func TestSyncReplicaSetDeletes(t *testing.T) {
client := clientset.NewForConfigOrDie(&restclient.Config{Host: "", ContentConfig: restclient.ContentConfig{GroupVersion: &api.Registry.GroupOrDie(v1.GroupName).GroupVersion}})
labelMap := map[string]string{"foo": "bar"}
rsSpec := newReplicaSet(1, labelMap)
client := fake.NewSimpleClientset(rsSpec)
fakePodControl := controller.FakePodControl{}
stopCh := make(chan struct{})
defer close(stopCh)
......@@ -255,8 +259,6 @@ func TestSyncReplicaSetDeletes(t *testing.T) {
manager.podControl = &fakePodControl
// 2 running pods and a controller with 1 replica, one pod delete expected
labelMap := map[string]string{"foo": "bar"}
rsSpec := newReplicaSet(1, labelMap)
informers.Extensions().V1beta1().ReplicaSets().Informer().GetIndexer().Add(rsSpec)
newPodList(informers.Core().V1().Pods().Informer().GetIndexer(), 2, v1.PodRunning, labelMap, rsSpec, "pod")
......@@ -300,15 +302,15 @@ func TestDeleteFinalStateUnknown(t *testing.T) {
}
func TestSyncReplicaSetCreates(t *testing.T) {
client := clientset.NewForConfigOrDie(&restclient.Config{Host: "", ContentConfig: restclient.ContentConfig{GroupVersion: &api.Registry.GroupOrDie(v1.GroupName).GroupVersion}})
labelMap := map[string]string{"foo": "bar"}
rs := newReplicaSet(2, labelMap)
client := fake.NewSimpleClientset(rs)
stopCh := make(chan struct{})
defer close(stopCh)
manager, informers := testNewReplicaSetControllerFromClient(client, stopCh, BurstReplicas)
// A controller with 2 replicas and no active pods in the store.
// Inactive pods should be ignored. 2 creates expected.
labelMap := map[string]string{"foo": "bar"}
rs := newReplicaSet(2, labelMap)
informers.Extensions().V1beta1().ReplicaSets().Informer().GetIndexer().Add(rs)
failedPod := newPod("failed-pod", rs, v1.PodFailed, nil, true)
deletedPod := newPod("deleted-pod", rs, v1.PodRunning, nil, true)
......@@ -778,21 +780,20 @@ func TestUpdatePods(t *testing.T) {
func TestControllerUpdateRequeue(t *testing.T) {
// This server should force a requeue of the controller because it fails to update status.Replicas.
fakeHandler := utiltesting.FakeHandler{
StatusCode: 500,
ResponseBody: "{}",
SkipRequestFn: skipListerFunc,
}
testServer := httptest.NewServer(&fakeHandler)
defer testServer.Close()
labelMap := map[string]string{"foo": "bar"}
rs := newReplicaSet(1, labelMap)
client := fake.NewSimpleClientset(rs)
client.PrependReactor("update", "replicasets",
func(action core.Action) (bool, runtime.Object, error) {
if action.GetSubresource() != "status" {
return false, nil, nil
}
return true, nil, errors.New("failed to update status")
})
stopCh := make(chan struct{})
defer close(stopCh)
client := clientset.NewForConfigOrDie(&restclient.Config{Host: testServer.URL, ContentConfig: restclient.ContentConfig{GroupVersion: &api.Registry.GroupOrDie(v1.GroupName).GroupVersion}})
manager, informers := testNewReplicaSetControllerFromClient(client, stopCh, BurstReplicas)
labelMap := map[string]string{"foo": "bar"}
rs := newReplicaSet(1, labelMap)
informers.Extensions().V1beta1().ReplicaSets().Informer().GetIndexer().Add(rs)
rs.Status = extensions.ReplicaSetStatus{Replicas: 2}
newPodList(informers.Core().V1().Pods().Informer().GetIndexer(), 1, v1.PodRunning, labelMap, rs, "pod")
......@@ -800,13 +801,14 @@ func TestControllerUpdateRequeue(t *testing.T) {
fakePodControl := controller.FakePodControl{}
manager.podControl = &fakePodControl
// an error from the sync function will be requeued, check to make sure we returned an error
err := manager.syncReplicaSet(getKey(rs, t))
if err == nil {
t.Errorf("missing error for requeue")
// Enqueue once. Then process it. Disable rate-limiting for this.
manager.queue = workqueue.NewRateLimitingQueue(workqueue.NewMaxOfRateLimiter())
manager.enqueueReplicaSet(rs)
manager.processNextWorkItem()
// It should have been requeued.
if got, want := manager.queue.Len(), 1; got != want {
t.Errorf("queue.Len() = %v, want %v", got, want)
}
// 1 Update and 1 GET, both of which fail
fakeHandler.ValidateRequestCount(t, 2)
}
func TestControllerUpdateStatusWithFailure(t *testing.T) {
......@@ -856,15 +858,15 @@ func TestControllerUpdateStatusWithFailure(t *testing.T) {
// TODO: This test is too hairy for a unittest. It should be moved to an E2E suite.
func doTestControllerBurstReplicas(t *testing.T, burstReplicas, numReplicas int) {
client := clientset.NewForConfigOrDie(&restclient.Config{Host: "", ContentConfig: restclient.ContentConfig{GroupVersion: &api.Registry.GroupOrDie(v1.GroupName).GroupVersion}})
labelMap := map[string]string{"foo": "bar"}
rsSpec := newReplicaSet(numReplicas, labelMap)
client := fake.NewSimpleClientset(rsSpec)
fakePodControl := controller.FakePodControl{}
stopCh := make(chan struct{})
defer close(stopCh)
manager, informers := testNewReplicaSetControllerFromClient(client, stopCh, burstReplicas)
manager.podControl = &fakePodControl
labelMap := map[string]string{"foo": "bar"}
rsSpec := newReplicaSet(numReplicas, labelMap)
informers.Extensions().V1beta1().ReplicaSets().Informer().GetIndexer().Add(rsSpec)
expectedPods := int32(0)
......@@ -1044,12 +1046,12 @@ func TestRSSyncExpectations(t *testing.T) {
}
func TestDeleteControllerAndExpectations(t *testing.T) {
client := clientset.NewForConfigOrDie(&restclient.Config{Host: "", ContentConfig: restclient.ContentConfig{GroupVersion: &api.Registry.GroupOrDie(v1.GroupName).GroupVersion}})
rs := newReplicaSet(1, map[string]string{"foo": "bar"})
client := fake.NewSimpleClientset(rs)
stopCh := make(chan struct{})
defer close(stopCh)
manager, informers := testNewReplicaSetControllerFromClient(client, stopCh, 10)
rs := newReplicaSet(1, map[string]string{"foo": "bar"})
informers.Extensions().V1beta1().ReplicaSets().Informer().GetIndexer().Add(rs)
fakePodControl := controller.FakePodControl{}
......@@ -1440,11 +1442,11 @@ func TestUpdateSelectorControllerRef(t *testing.T) {
func TestDoNotAdoptOrCreateIfBeingDeleted(t *testing.T) {
labelMap := map[string]string{"foo": "bar"}
rs := newReplicaSet(2, labelMap)
now := metav1.Now()
rs.DeletionTimestamp = &now
stopCh := make(chan struct{})
defer close(stopCh)
manager, fakePodControl, informers := setupManagerWithGCEnabled(stopCh, rs)
now := metav1.Now()
rs.DeletionTimestamp = &now
informers.Extensions().V1beta1().ReplicaSets().Informer().GetIndexer().Add(rs)
pod1 := newPod("pod1", rs, v1.PodRunning, nil, false)
informers.Core().V1().Pods().Informer().GetIndexer().Add(pod1)
......@@ -1457,6 +1459,33 @@ func TestDoNotAdoptOrCreateIfBeingDeleted(t *testing.T) {
validateSyncReplicaSet(t, fakePodControl, 0, 0, 0)
}
func TestDoNotAdoptOrCreateIfBeingDeletedRace(t *testing.T) {
labelMap := map[string]string{"foo": "bar"}
// Bare client says it IS deleted.
rs := newReplicaSet(2, labelMap)
now := metav1.Now()
rs.DeletionTimestamp = &now
stopCh := make(chan struct{})
defer close(stopCh)
manager, fakePodControl, informers := setupManagerWithGCEnabled(stopCh, rs)
// Lister (cache) says it's NOT deleted.
rs2 := *rs
rs2.DeletionTimestamp = nil
informers.Extensions().V1beta1().ReplicaSets().Informer().GetIndexer().Add(&rs2)
// Recheck occurs if a matching orphan is present.
pod1 := newPod("pod1", rs, v1.PodRunning, nil, false)
informers.Core().V1().Pods().Informer().GetIndexer().Add(pod1)
// sync should abort.
err := manager.syncReplicaSet(getKey(rs, t))
if err == nil {
t.Error("syncReplicaSet() err = nil, expected non-nil")
}
// no patch, no create.
validateSyncReplicaSet(t, fakePodControl, 0, 0, 0)
}
func TestReadyReplicas(t *testing.T) {
// This is a happy server just to record the PUT request we expect for status.Replicas
fakeHandler := utiltesting.FakeHandler{
......
pkg/controller/replication/BUILD
View file @ 47320fd3
......@@ -66,6 +66,7 @@ go_test(
"//vendor:k8s.io/client-go/testing",
"//vendor:k8s.io/client-go/tools/cache",
"//vendor:k8s.io/client-go/util/testing",
"//vendor:k8s.io/client-go/util/workqueue",
],
)
......
pkg/controller/replication/replication_controller.go
View file @ 47320fd3
......@@ -582,7 +582,19 @@ func (rm *ReplicationManager) syncReplicationController(key string) error {
filteredPods = append(filteredPods, pod)
}
}
cm := controller.NewPodControllerRefManager(rm.podControl, rc, labels.Set(rc.Spec.Selector).AsSelectorPreValidated(), controllerKind)
// If any adoptions are attempted, we should first recheck for deletion with
// an uncached quorum read sometime after listing Pods (see #42639).
canAdoptFunc := controller.RecheckDeletionTimestamp(func() (metav1.Object, error) {
fresh, err := rm.kubeClient.CoreV1().ReplicationControllers(rc.Namespace).Get(rc.Name, metav1.GetOptions{})
if err != nil {
return nil, err
}
if fresh.UID != rc.UID {
return nil, fmt.Errorf("original ReplicationController %v/%v is gone: got uid %v, wanted %v", rc.Namespace, rc.Name, fresh.UID, rc.UID)
}
return fresh, nil
})
cm := controller.NewPodControllerRefManager(rm.podControl, rc, labels.Set(rc.Spec.Selector).AsSelectorPreValidated(), controllerKind, canAdoptFunc)
// NOTE: filteredPods are pointing to objects from cache - if you need to
// modify them, you need to copy it first.
filteredPods, err = cm.ClaimPods(filteredPods)
......
pkg/controller/statefulset/stateful_set.go
View file @ 47320fd3
......@@ -289,7 +289,20 @@ func (ssc *StatefulSetController) getPodsForStatefulSet(set *apps.StatefulSet, s
return isMemberOf(set, pod)
}
cm := controller.NewPodControllerRefManager(ssc.podControl, set, selector, controllerKind)
// If any adoptions are attempted, we should first recheck for deletion with
// an uncached quorum read sometime after listing Pods (see #42639).
canAdoptFunc := controller.RecheckDeletionTimestamp(func() (metav1.Object, error) {
fresh, err := ssc.kubeClient.AppsV1beta1().StatefulSets(set.Namespace).Get(set.Name, metav1.GetOptions{})
if err != nil {
return nil, err
}
if fresh.UID != set.UID {
return nil, fmt.Errorf("original StatefulSet %v/%v is gone: got uid %v, wanted %v", set.Namespace, set.Name, fresh.UID, set.UID)
}
return fresh, nil
})
cm := controller.NewPodControllerRefManager(ssc.podControl, set, selector, controllerKind, canAdoptFunc)
return cm.ClaimPods(pods, filter)
}
......@@ -386,6 +399,7 @@ func (ssc *StatefulSetController) sync(key string) error {
utilruntime.HandleError(fmt.Errorf("unable to retrieve StatefulSet %v from store: %v", key, err))
return err
}
selector, err := metav1.LabelSelectorAsSelector(set.Spec.Selector)
if err != nil {
utilruntime.HandleError(fmt.Errorf("error converting StatefulSet %v selector: %v", key, err))
......
pkg/controller/statefulset/stateful_set_control_test.go
View file @ 47320fd3
......@@ -214,69 +214,6 @@ func TestStatefulSetControlReplacesPods(t *testing.T) {
}
}
func TestStatefulSetDeletionTimestamp(t *testing.T) {
set := newStatefulSet(5)
client := fake.NewSimpleClientset(set)
informerFactory := informers.NewSharedInformerFactory(client, controller.NoResyncPeriodFunc())
spc := newFakeStatefulPodControl(informerFactory.Core().V1().Pods(), informerFactory.Apps().V1beta1().StatefulSets())
ssc := NewDefaultStatefulSetControl(spc)
stop := make(chan struct{})
defer close(stop)
informerFactory.Start(stop)
cache.WaitForCacheSync(
stop,
informerFactory.Apps().V1beta1().StatefulSets().Informer().HasSynced,
informerFactory.Core().V1().Pods().Informer().HasSynced,
)
// Bring up a StatefulSet.
if err := scaleUpStatefulSetControl(set, ssc, spc); err != nil {
t.Errorf("failed to turn up StatefulSet : %s", err)
}
var err error
set, err = spc.setsLister.StatefulSets(set.Namespace).Get(set.Name)
if err != nil {
t.Fatalf("error getting updated StatefulSet: %v", err)
}
if set.Status.Replicas != 5 {
t.Error("failed to scale statefulset to 5 replicas")
}
selector, err := metav1.LabelSelectorAsSelector(set.Spec.Selector)
if err != nil {
t.Error(err)
}
pods, err := spc.podsLister.Pods(set.Namespace).List(selector)
if err != nil {
t.Error(err)
}
sort.Sort(ascendingOrdinal(pods))
// Mark the StatefulSet as being deleted.
set.DeletionTimestamp = new(metav1.Time)
// Delete the first pod.
spc.podsIndexer.Delete(pods[0])
pods, err = spc.podsLister.Pods(set.Namespace).List(selector)
if err != nil {
t.Error(err)
}
// The StatefulSet should update its replica count,
// but not try to fix it.
if err := ssc.UpdateStatefulSet(set, pods); err != nil {
t.Errorf("failed to update StatefulSet : %s", err)
}
set, err = spc.setsLister.StatefulSets(set.Namespace).Get(set.Name)
if err != nil {
t.Fatalf("error getting updated StatefulSet: %v", err)
}
if e, a := int32(4), set.Status.Replicas; e != a {
t.Errorf("expected to scale to %d, got %d", e, a)
}
}
func TestDefaultStatefulSetControlRecreatesFailedPod(t *testing.T) {
client := fake.NewSimpleClientset()
informerFactory := informers.NewSharedInformerFactory(client, controller.NoResyncPeriodFunc())
......
pkg/controller/statefulset/stateful_set_test.go
View file @ 47320fd3
......@@ -22,9 +22,8 @@ import (
"testing"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/client-go/tools/cache"
"k8s.io/kubernetes/pkg/api/v1"
apps "k8s.io/kubernetes/pkg/apis/apps/v1beta1"
"k8s.io/kubernetes/pkg/client/clientset_generated/clientset/fake"
......@@ -35,8 +34,8 @@ import (
func alwaysReady() bool { return true }
func TestStatefulSetControllerCreates(t *testing.T) {
ssc, spc := newFakeStatefulSetController()
set := newStatefulSet(3)
ssc, spc := newFakeStatefulSetController(set)
if err := scaleUpStatefulSetController(set, ssc, spc); err != nil {
t.Errorf("Failed to turn up StatefulSet : %s", err)
}
......@@ -51,8 +50,8 @@ func TestStatefulSetControllerCreates(t *testing.T) {
}
func TestStatefulSetControllerDeletes(t *testing.T) {
ssc, spc := newFakeStatefulSetController()
set := newStatefulSet(3)
ssc, spc := newFakeStatefulSetController(set)
if err := scaleUpStatefulSetController(set, ssc, spc); err != nil {
t.Errorf("Failed to turn up StatefulSet : %s", err)
}
......@@ -79,8 +78,8 @@ func TestStatefulSetControllerDeletes(t *testing.T) {
}
func TestStatefulSetControllerRespectsTermination(t *testing.T) {
ssc, spc := newFakeStatefulSetController()
set := newStatefulSet(3)
ssc, spc := newFakeStatefulSetController(set)
if err := scaleUpStatefulSetController(set, ssc, spc); err != nil {
t.Errorf("Failed to turn up StatefulSet : %s", err)
}
......@@ -130,8 +129,8 @@ func TestStatefulSetControllerRespectsTermination(t *testing.T) {
}
func TestStatefulSetControllerBlocksScaling(t *testing.T) {
ssc, spc := newFakeStatefulSetController()
set := newStatefulSet(3)
ssc, spc := newFakeStatefulSetController(set)
if err := scaleUpStatefulSetController(set, ssc, spc); err != nil {
t.Errorf("Failed to turn up StatefulSet : %s", err)
}
......@@ -176,6 +175,63 @@ func TestStatefulSetControllerBlocksScaling(t *testing.T) {
}
}
func TestStatefulSetControllerDeletionTimestamp(t *testing.T) {
set := newStatefulSet(3)
set.DeletionTimestamp = new(metav1.Time)
ssc, spc := newFakeStatefulSetController(set)
spc.setsIndexer.Add(set)
// Force a sync. It should not try to create any Pods.
ssc.enqueueStatefulSet(set)
fakeWorker(ssc)
selector, err := metav1.LabelSelectorAsSelector(set.Spec.Selector)
if err != nil {
t.Fatal(err)
}
pods, err := spc.podsLister.Pods(set.Namespace).List(selector)
if err != nil {
t.Fatal(err)
}
if got, want := len(pods), 0; got != want {
t.Errorf("len(pods) = %v, want %v", got, want)
}
}
func TestStatefulSetControllerDeletionTimestampRace(t *testing.T) {
set := newStatefulSet(3)
// The bare client says it IS deleted.
set.DeletionTimestamp = new(metav1.Time)
ssc, spc := newFakeStatefulSetController(set)
// The lister (cache) says it's NOT deleted.
set2 := *set
set2.DeletionTimestamp = nil
spc.setsIndexer.Add(&set2)
// The recheck occurs in the presence of a matching orphan.
pod := newStatefulSetPod(set, 1)
pod.OwnerReferences = nil
spc.podsIndexer.Add(pod)
// Force a sync. It should not try to create any Pods.
ssc.enqueueStatefulSet(set)
fakeWorker(ssc)
selector, err := metav1.LabelSelectorAsSelector(set.Spec.Selector)
if err != nil {
t.Fatal(err)
}
pods, err := spc.podsLister.Pods(set.Namespace).List(selector)
if err != nil {
t.Fatal(err)
}
if got, want := len(pods), 1; got != want {
t.Errorf("len(pods) = %v, want %v", got, want)
}
}
func TestStatefulSetControllerAddPod(t *testing.T) {
ssc, spc := newFakeStatefulSetController()
set1 := newStatefulSet(3)
......@@ -437,8 +493,8 @@ func TestStatefulSetControllerGetStatefulSetsForPod(t *testing.T) {
}
func TestGetPodsForStatefulSetAdopt(t *testing.T) {
ssc, spc := newFakeStatefulSetController()
set := newStatefulSet(5)
ssc, spc := newFakeStatefulSetController(set)
pod1 := newStatefulSetPod(set, 1)
// pod2 is an orphan with matching labels and name.
pod2 := newStatefulSetPod(set, 2)
......@@ -479,8 +535,8 @@ func TestGetPodsForStatefulSetAdopt(t *testing.T) {
}
func TestGetPodsForStatefulSetRelease(t *testing.T) {
ssc, spc := newFakeStatefulSetController()
set := newStatefulSet(3)
ssc, spc := newFakeStatefulSetController(set)
pod1 := newStatefulSetPod(set, 1)
// pod2 is owned but has wrong name.
pod2 := newStatefulSetPod(set, 2)
......@@ -518,8 +574,8 @@ func TestGetPodsForStatefulSetRelease(t *testing.T) {
}
}
func newFakeStatefulSetController() (*StatefulSetController, *fakeStatefulPodControl) {
client := fake.NewSimpleClientset()
func newFakeStatefulSetController(initialObjects ...runtime.Object) (*StatefulSetController, *fakeStatefulPodControl) {
client := fake.NewSimpleClientset(initialObjects...)
informerFactory := informers.NewSharedInformerFactory(client, controller.NoResyncPeriodFunc())
fpc := newFakeStatefulPodControl(informerFactory.Core().V1().Pods(), informerFactory.Apps().V1beta1().StatefulSets())
ssc := NewStatefulSetController(
......
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