
Automatic merge from submit-queue volume controller: Add cache with the latest version of PVs and PVCs When the controller binds a PV to PVC, it saves both objects to etcd. However, there is still an old version of these objects in the controller Informer cache. So, when a new PVC comes, the PV is still seen as available and may get bound to the new PVC. This will be blocked by etcd, still, it creates unnecessary traffic that slows everything down. To make everything worse, when periodic sync with the old PVC is performed, this PVC is seen by the controller as Pending (while it's already Bound on etcd) and will be bound to a different PV. Writing to this PV won't be blocked by etcd, only subsequent write of the PVC fails. So, the controller will need to roll back the PV in another transaction(s). The controller can keep itself pretty busy this way. Also, we save bound PVs (and PVCs) as two transactions - we save say PV.Spec first and then .Status. The controller gets "PV.Spec updated" event from etcd and tries to fix the Status, as it seems to the controller it's outdated. This write again fails - there already is a correct version in etcd. As we can't influence the Informer cache, it is read-only to the controller, this patch introduces second cache in the controller, which holds latest and greatest version on PVs and PVCs to prevent these useless writes to etcd . It gets updated with events from etcd *and* after etcd confirms successful save of PV/PVC modified by the controller. The cache stores only *pointers* to PVs/PVCs, so in ideal case it shares the actual object data with the informer cache. They will diverge only for a short time when the controller modifies something and the informer cache did not get update events yet. @kubernetes/sig-storage
236 lines
8.9 KiB
Go
236 lines
8.9 KiB
Go
/*
|
|
Copyright 2016 The Kubernetes Authors All rights reserved.
|
|
|
|
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 persistentvolume
|
|
|
|
import (
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/golang/glog"
|
|
"k8s.io/kubernetes/pkg/api"
|
|
"k8s.io/kubernetes/pkg/client/cache"
|
|
"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/fake"
|
|
"k8s.io/kubernetes/pkg/controller/framework"
|
|
"k8s.io/kubernetes/pkg/conversion"
|
|
)
|
|
|
|
// Test the real controller methods (add/update/delete claim/volume) with
|
|
// a fake API server.
|
|
// There is no controller API to 'initiate syncAll now', therefore these tests
|
|
// can't reliably simulate periodic sync of volumes/claims - it would be
|
|
// either very timing-sensitive or slow to wait for real periodic sync.
|
|
func TestControllerSync(t *testing.T) {
|
|
expectedChanges := []int{1, 4, 1, 1}
|
|
tests := []controllerTest{
|
|
// [Unit test set 5] - controller tests.
|
|
// We test the controller as if
|
|
// it was connected to real API server, i.e. we call add/update/delete
|
|
// Claim/Volume methods. Also, all changes to volumes and claims are
|
|
// sent to add/update/delete Claim/Volume as real controller would do.
|
|
{
|
|
// addVolume gets a new volume. Check it's marked as Available and
|
|
// that it's not bound to any claim - we bind volumes on periodic
|
|
// syncClaim, not on addVolume.
|
|
"5-1 - addVolume",
|
|
novolumes, /* added in testCall below */
|
|
newVolumeArray("volume5-1", "10Gi", "", "", api.VolumeAvailable, api.PersistentVolumeReclaimRetain),
|
|
newClaimArray("claim5-1", "uid5-1", "1Gi", "", api.ClaimPending),
|
|
newClaimArray("claim5-1", "uid5-1", "1Gi", "", api.ClaimPending),
|
|
noevents, noerrors,
|
|
// Custom test function that generates an add event
|
|
func(ctrl *PersistentVolumeController, reactor *volumeReactor, test controllerTest) error {
|
|
volume := newVolume("volume5-1", "10Gi", "", "", api.VolumePending, api.PersistentVolumeReclaimRetain)
|
|
reactor.volumes[volume.Name] = volume
|
|
reactor.volumeSource.Add(volume)
|
|
return nil
|
|
},
|
|
},
|
|
{
|
|
// addClaim gets a new claim. Check it's bound to a volume.
|
|
"5-2 - complete bind",
|
|
newVolumeArray("volume5-2", "10Gi", "", "", api.VolumeAvailable, api.PersistentVolumeReclaimRetain),
|
|
newVolumeArray("volume5-2", "10Gi", "uid5-2", "claim5-2", api.VolumeBound, api.PersistentVolumeReclaimRetain, annBoundByController),
|
|
noclaims, /* added in testAddClaim5_2 */
|
|
newClaimArray("claim5-2", "uid5-2", "1Gi", "volume5-2", api.ClaimBound, annBoundByController, annBindCompleted),
|
|
noevents, noerrors,
|
|
// Custom test function that generates an add event
|
|
func(ctrl *PersistentVolumeController, reactor *volumeReactor, test controllerTest) error {
|
|
claim := newClaim("claim5-2", "uid5-2", "1Gi", "", api.ClaimPending)
|
|
reactor.claims[claim.Name] = claim
|
|
reactor.claimSource.Add(claim)
|
|
return nil
|
|
},
|
|
},
|
|
{
|
|
// deleteClaim with a bound claim makes bound volume released.
|
|
"5-3 - delete claim",
|
|
newVolumeArray("volume5-3", "10Gi", "uid5-3", "claim5-3", api.VolumeBound, api.PersistentVolumeReclaimRetain, annBoundByController),
|
|
newVolumeArray("volume5-3", "10Gi", "uid5-3", "claim5-3", api.VolumeReleased, api.PersistentVolumeReclaimRetain, annBoundByController),
|
|
newClaimArray("claim5-3", "uid5-3", "1Gi", "volume5-3", api.ClaimBound, annBoundByController, annBindCompleted),
|
|
noclaims,
|
|
noevents, noerrors,
|
|
// Custom test function that generates a delete event
|
|
func(ctrl *PersistentVolumeController, reactor *volumeReactor, test controllerTest) error {
|
|
obj := ctrl.claims.List()[0]
|
|
claim := obj.(*api.PersistentVolumeClaim)
|
|
// Remove the claim from list of resulting claims.
|
|
delete(reactor.claims, claim.Name)
|
|
// Poke the controller with deletion event. Cloned claim is
|
|
// needed to prevent races (and we would get a clone from etcd
|
|
// too).
|
|
clone, _ := conversion.NewCloner().DeepCopy(claim)
|
|
claimClone := clone.(*api.PersistentVolumeClaim)
|
|
reactor.claimSource.Delete(claimClone)
|
|
return nil
|
|
},
|
|
},
|
|
{
|
|
// deleteVolume with a bound volume. Check the claim is Lost.
|
|
"5-4 - delete volume",
|
|
newVolumeArray("volume5-4", "10Gi", "uid5-4", "claim5-4", api.VolumeBound, api.PersistentVolumeReclaimRetain),
|
|
novolumes,
|
|
newClaimArray("claim5-4", "uid5-4", "1Gi", "volume5-4", api.ClaimBound, annBoundByController, annBindCompleted),
|
|
newClaimArray("claim5-4", "uid5-4", "1Gi", "volume5-4", api.ClaimLost, annBoundByController, annBindCompleted),
|
|
[]string{"Warning ClaimLost"}, noerrors,
|
|
// Custom test function that generates a delete event
|
|
func(ctrl *PersistentVolumeController, reactor *volumeReactor, test controllerTest) error {
|
|
obj := ctrl.volumes.store.List()[0]
|
|
volume := obj.(*api.PersistentVolume)
|
|
// Remove the volume from list of resulting volumes.
|
|
delete(reactor.volumes, volume.Name)
|
|
// Poke the controller with deletion event. Cloned volume is
|
|
// needed to prevent races (and we would get a clone from etcd
|
|
// too).
|
|
clone, _ := conversion.NewCloner().DeepCopy(volume)
|
|
volumeClone := clone.(*api.PersistentVolume)
|
|
reactor.volumeSource.Delete(volumeClone)
|
|
return nil
|
|
},
|
|
},
|
|
}
|
|
|
|
for ix, test := range tests {
|
|
glog.V(4).Infof("starting test %q", test.name)
|
|
|
|
// Initialize the controller
|
|
client := &fake.Clientset{}
|
|
volumeSource := framework.NewFakeControllerSource()
|
|
claimSource := framework.NewFakeControllerSource()
|
|
ctrl := newTestController(client, volumeSource, claimSource)
|
|
reactor := newVolumeReactor(client, ctrl, volumeSource, claimSource, test.errors)
|
|
for _, claim := range test.initialClaims {
|
|
claimSource.Add(claim)
|
|
reactor.claims[claim.Name] = claim
|
|
}
|
|
for _, volume := range test.initialVolumes {
|
|
volumeSource.Add(volume)
|
|
reactor.volumes[volume.Name] = volume
|
|
}
|
|
|
|
// Start the controller
|
|
defer ctrl.Stop()
|
|
go ctrl.Run()
|
|
|
|
// Wait for the controller to pass initial sync.
|
|
for !ctrl.isFullySynced() {
|
|
time.Sleep(10 * time.Millisecond)
|
|
}
|
|
|
|
count := reactor.getChangeCount()
|
|
|
|
// Call the tested function
|
|
err := test.test(ctrl, reactor, test)
|
|
if err != nil {
|
|
t.Errorf("Test %q initial test call failed: %v", test.name, err)
|
|
}
|
|
|
|
for reactor.getChangeCount() < count+expectedChanges[ix] {
|
|
reactor.waitTest()
|
|
}
|
|
|
|
evaluateTestResults(ctrl, reactor, test, t)
|
|
}
|
|
}
|
|
|
|
func storeVersion(t *testing.T, prefix string, c cache.Store, version string, expectedReturn bool) {
|
|
pv := newVolume("pvName", "1Gi", "", "", api.VolumeAvailable, api.PersistentVolumeReclaimDelete)
|
|
pv.ResourceVersion = version
|
|
ret, err := storeObjectUpdate(c, pv, "volume")
|
|
if err != nil {
|
|
t.Errorf("%s: expected storeObjectUpdate to succeed, got: %v", prefix, err)
|
|
}
|
|
if expectedReturn != ret {
|
|
t.Errorf("%s: expected storeObjectUpdate to return %v, got: %v", prefix, expectedReturn, ret)
|
|
}
|
|
|
|
// find the stored version
|
|
|
|
pvObj, found, err := c.GetByKey("pvName")
|
|
if err != nil {
|
|
t.Errorf("expected volume 'pvName' in the cache, got error instead: %v", err)
|
|
}
|
|
if !found {
|
|
t.Errorf("expected volume 'pvName' in the cache but it was not found")
|
|
}
|
|
pv, ok := pvObj.(*api.PersistentVolume)
|
|
if !ok {
|
|
t.Errorf("expected volume in the cache, got different object instead: %+v", pvObj)
|
|
}
|
|
|
|
if ret {
|
|
if pv.ResourceVersion != version {
|
|
t.Errorf("expected volume with version %s in the cache, got %s instead", version, pv.ResourceVersion)
|
|
}
|
|
} else {
|
|
if pv.ResourceVersion == version {
|
|
t.Errorf("expected volume with version other than %s in the cache, got %s instead", version, pv.ResourceVersion)
|
|
}
|
|
}
|
|
}
|
|
|
|
// TestControllerCache tests func storeObjectUpdate()
|
|
func TestControllerCache(t *testing.T) {
|
|
// Cache under test
|
|
c := cache.NewStore(framework.DeletionHandlingMetaNamespaceKeyFunc)
|
|
|
|
// Store new PV
|
|
storeVersion(t, "Step1", c, "1", true)
|
|
// Store the same PV
|
|
storeVersion(t, "Step2", c, "1", true)
|
|
// Store newer PV
|
|
storeVersion(t, "Step3", c, "2", true)
|
|
// Store older PV - simulating old "PV updated" event or periodic sync with
|
|
// old data
|
|
storeVersion(t, "Step4", c, "1", false)
|
|
// Store newer PV - test integer parsing ("2" > "10" as string,
|
|
// while 2 < 10 as integers)
|
|
storeVersion(t, "Step5", c, "10", true)
|
|
}
|
|
|
|
func TestControllerCacheParsingError(t *testing.T) {
|
|
c := cache.NewStore(framework.DeletionHandlingMetaNamespaceKeyFunc)
|
|
// There must be something in the cache to compare with
|
|
storeVersion(t, "Step1", c, "1", true)
|
|
|
|
pv := newVolume("pvName", "1Gi", "", "", api.VolumeAvailable, api.PersistentVolumeReclaimDelete)
|
|
pv.ResourceVersion = "xxx"
|
|
_, err := storeObjectUpdate(c, pv, "volume")
|
|
if err == nil {
|
|
t.Errorf("Expected parsing error, got nil instead")
|
|
}
|
|
}
|