Fixed persistent volume claim controllers processing an old volume

We should always load the newest version of the volume from APIserver before
processing it.

When PersistentVolumeProvisionerController.reconcileVolume() is called with the
same volume in short succession (e.g. the volume is created by a provisioner
and at the same time periodic check of all volumes is scheduled), the second
reconcileVolume() call gets an old copy of the volume as its parameter and
it does not see annotations updated by the previous call.

This may result in one volume being provisioned several times, creating orphan
volumes in the cloud.

The same error is in PersistentVolumeClaimBinder.syncVolume().

pkg/controller/persistentvolume/persistentvolume_claim_binder_controller.go

@@ -193,6 +193,14 @@ func (binder *PersistentVolumeClaimBinder) deleteClaim(obj interface{}) {
 func syncVolume(volumeIndex *persistentVolumeOrderedIndex, binderClient binderClient, volume *api.PersistentVolume) (err error) {
 	glog.V(5).Infof("Synchronizing PersistentVolume[%s], current phase: %s\n", volume.Name, volume.Status.Phase)
 
+	// The PV may have been modified by parallel call to syncVolume, load
+	// the current version.
+	newPv, err := binderClient.GetPersistentVolume(volume.Name)
+	if err != nil {
+		return fmt.Errorf("Cannot reload volume %s: %v", volume.Name, err)
+	}
+	volume = newPv
+
 	// volumes can be in one of the following states:
 	//
 	// VolumePending -- default value -- not bound to a claim and not yet processed through this controller.
@@ -203,12 +211,15 @@ func syncVolume(volumeIndex *persistentVolumeOrderedIndex, binderClient binderCl
 	currentPhase := volume.Status.Phase
 	nextPhase := currentPhase
 
+	// Always store the newest volume state in local cache.
 	_, exists, err := volumeIndex.Get(volume)
 	if err != nil {
 		return err
 	}
 	if !exists {
 		volumeIndex.Add(volume)
+	} else {
+		volumeIndex.Update(volume)
 	}
 
 	if isBeingProvisioned(volume) {

pkg/controller/persistentvolume/persistentvolume_claim_binder_controller_test.go

@@ -120,6 +120,7 @@ func TestClaimRace(t *testing.T) {
 
 	volumeIndex := NewPersistentVolumeOrderedIndex()
 	mockClient := &mockBinderClient{}
+	mockClient.volume = v
 
 	plugMgr := volume.VolumePluginMgr{}
 	plugMgr.InitPlugins(host_path.ProbeRecyclableVolumePlugins(newMockRecycler, volume.VolumeConfig{}), volume.NewFakeVolumeHost(tmpDir, nil, nil))
@@ -211,7 +212,8 @@ func TestClaimSyncAfterVolumeProvisioning(t *testing.T) {
 
 	volumeIndex := NewPersistentVolumeOrderedIndex()
 	mockClient := &mockBinderClient{
-		claim: claim,
+		claim:  claim,
+		volume: pv,
 	}
 
 	plugMgr := volume.VolumePluginMgr{}

pkg/controller/persistentvolume/persistentvolume_provisioner_controller.go

@@ -211,6 +211,14 @@ func (controller *PersistentVolumeProvisionerController) reconcileClaim(claim *a
 func (controller *PersistentVolumeProvisionerController) reconcileVolume(pv *api.PersistentVolume) error {
 	glog.V(5).Infof("PersistentVolume[%s] reconciling", pv.Name)
 
+	// The PV may have been modified by parallel call to reconcileVolume, load
+	// the current version.
+	newPv, err := controller.client.GetPersistentVolume(pv.Name)
+	if err != nil {
+		return fmt.Errorf("Cannot reload volume %s: %v", pv.Name, err)
+	}
+	pv = newPv
+
 	if pv.Spec.ClaimRef == nil {
 		glog.V(5).Infof("PersistentVolume[%s] is not bound to a claim.  No provisioning required", pv.Name)
 		return nil
@@ -244,7 +252,7 @@ func (controller *PersistentVolumeProvisionerController) reconcileVolume(pv *api
 
 	// provisioning is incomplete.  Attempt to provision the volume.
 	glog.V(5).Infof("PersistentVolume[%s] provisioning in progress", pv.Name)
-	err := provisionVolume(pv, controller)
+	err = provisionVolume(pv, controller)
 	if err != nil {
 		return fmt.Errorf("Error provisioning PersistentVolume[%s]: %v", err)
 	}

pkg/controller/persistentvolume/persistentvolume_provisioner_controller_test.go

@@ -144,6 +144,7 @@ func TestReconcileVolume(t *testing.T) {
 	controller, mockClient, mockVolumePlugin := makeTestController()
 	pv := makeTestVolume()
 	pvc := makeTestClaim()
+	mockClient.volume = pv
 
 	err := controller.reconcileVolume(pv)
 	if err != nil {
@@ -158,6 +159,7 @@ func TestReconcileVolume(t *testing.T) {
 	// pretend the claim and volume are bound, no provisioning required
 	claimRef, _ := api.GetReference(pvc)
 	pv.Spec.ClaimRef = claimRef
+	mockClient.volume = pv
 	err = controller.reconcileVolume(pv)
 	if err != nil {
 		t.Errorf("Unexpected error %v", err)
@@ -165,6 +167,7 @@ func TestReconcileVolume(t *testing.T) {
 
 	pv.Annotations[pvProvisioningRequiredAnnotationKey] = "!pvProvisioningCompleted"
 	pv.Annotations[qosProvisioningKey] = "foo"
+	mockClient.volume = pv
 	err = controller.reconcileVolume(pv)
 
 	if !isAnnotationMatch(pvProvisioningRequiredAnnotationKey, pvProvisioningCompletedAnnotationValue, mockClient.volume.Annotations) {