Merge pull request #16432 from markturansky/recycler_race

pkg/controller/persistentvolume/persistentvolume_claim_binder_controller.go

@@ -27,6 +27,7 @@ import (
 	"k8s.io/kubernetes/pkg/client/cache"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/controller/framework"
+	"k8s.io/kubernetes/pkg/conversion"
 	"k8s.io/kubernetes/pkg/fields"
 	"k8s.io/kubernetes/pkg/labels"
 	"k8s.io/kubernetes/pkg/runtime"
@@ -159,44 +160,53 @@ func syncVolume(volumeIndex *persistentVolumeOrderedIndex, binderClient binderCl
 	currentPhase := volume.Status.Phase
 	nextPhase := currentPhase
 
+	_, exists, err := volumeIndex.Get(volume)
+	if err != nil {
+		return err
+	}
+	if !exists {
+		volumeIndex.Add(volume)
+	}
+
 	switch currentPhase {
-	// pending volumes are available only after indexing in order to be matched to claims.
 	case api.VolumePending:
+
+		// 3 possible states:
+		//  1.  ClaimRef != nil and Claim exists:   Prebound to claim. Make volume available for binding (it will match PVC).
+		//  2.  ClaimRef != nil and Claim !exists:  Recently recycled. Remove bind. Make volume available for new claim.
+		//  3.  ClaimRef == nil: Neither recycled nor prebound.  Make volume available for binding.
+		nextPhase = api.VolumeAvailable
+
 		if volume.Spec.ClaimRef != nil {
-			// Pending volumes that have a ClaimRef were recently recycled.  The Recycler set the phase to VolumePending
-			// to start the volume again at the beginning of this lifecycle.
-			// ClaimRef is the last bind between persistent volume and claim.
-			// The claim has already been deleted by the user at this point
-			oldClaimRef := volume.Spec.ClaimRef
-			volume.Spec.ClaimRef = nil
-			_, err = binderClient.UpdatePersistentVolume(volume)
-			if err != nil {
-				// rollback on error, keep the ClaimRef until we can successfully update the volume
-				volume.Spec.ClaimRef = oldClaimRef
-				return fmt.Errorf("Unexpected error saving PersistentVolume: %+v", err)
-			}
-		}
+			_, err := binderClient.GetPersistentVolumeClaim(volume.Spec.ClaimRef.Namespace, volume.Spec.ClaimRef.Name)
+			if errors.IsNotFound(err) {
+				// Pending volumes that have a ClaimRef where the claim is missing were recently recycled.
+				// The Recycler set the phase to VolumePending to start the volume at the beginning of this lifecycle.
+				// removing ClaimRef unbinds the volume
+				clone, err := conversion.NewCloner().DeepCopy(volume)
+				if err != nil {
+					return fmt.Errorf("Error cloning pv: %v", err)
+				}
+				volumeClone, ok := clone.(*api.PersistentVolume)
+				if !ok {
+					return fmt.Errorf("Unexpected pv cast error : %v\n", volumeClone)
+				}
+				volumeClone.Spec.ClaimRef = nil
 
-		_, exists, err := volumeIndex.Get(volume)
-		if err != nil {
-			return err
-		}
-		if !exists {
-			volumeIndex.Add(volume)
+				if updatedVolume, err := binderClient.UpdatePersistentVolume(volumeClone); err != nil {
+					return fmt.Errorf("Unexpected error saving PersistentVolume: %+v", err)
+				} else {
+					volume = updatedVolume
+					volumeIndex.Update(volume)
+				}
+			} else if err != nil {
+				return fmt.Errorf("Error getting PersistentVolumeClaim[%s/%s]: %v", volume.Spec.ClaimRef.Namespace, volume.Spec.ClaimRef.Name, err)
+			}
 		}
-		glog.V(5).Infof("PersistentVolume[%s] is now available\n", volume.Name)
-		nextPhase = api.VolumeAvailable
+		glog.V(5).Infof("PersistentVolume[%s] is available\n", volume.Name)
 
 	// available volumes await a claim
 	case api.VolumeAvailable:
-		// TODO:  remove api.VolumePending phase altogether
-		_, exists, err := volumeIndex.Get(volume)
-		if err != nil {
-			return err
-		}
-		if !exists {
-			volumeIndex.Add(volume)
-		}
 		if volume.Spec.ClaimRef != nil {
 			_, err := binderClient.GetPersistentVolumeClaim(volume.Spec.ClaimRef.Namespace, volume.Spec.ClaimRef.Name)
 			if err == nil {
@@ -265,79 +275,71 @@ func syncVolume(volumeIndex *persistentVolumeOrderedIndex, binderClient binderCl
 func syncClaim(volumeIndex *persistentVolumeOrderedIndex, binderClient binderClient, claim *api.PersistentVolumeClaim) (err error) {
 	glog.V(5).Infof("Synchronizing PersistentVolumeClaim[%s]\n", claim.Name)
 
-	// claims can be in one of the following states:
-	//
-	// ClaimPending -- default value -- not bound to a claim.  A volume that matches the claim may not exist.
-	// ClaimBound -- bound to a volume.  claim.Status.VolumeRef != nil
-	currentPhase := claim.Status.Phase
-	nextPhase := currentPhase
-
-	switch currentPhase {
-	// pending claims await a matching volume
+	switch claim.Status.Phase {
 	case api.ClaimPending:
-		volume, err := volumeIndex.FindBestMatchForClaim(claim)
+		volume, err := volumeIndex.findBestMatchForClaim(claim)
 		if err != nil {
 			return err
 		}
 		if volume == nil {
-			return fmt.Errorf("A volume match does not exist for persistent claim: %s", claim.Name)
+			glog.V(5).Infof("A volume match does not exist for persistent claim: %s", claim.Name)
+			return nil
 		}
 
-		// make a binding reference to the claim.
-		// triggers update of the claim in this controller, which builds claim status
-		claim.Spec.VolumeName = volume.Name
-		// TODO: make this similar to Pod's binding both with BindingREST subresource and GuaranteedUpdate helper in etcd.go
-		claim, err = binderClient.UpdatePersistentVolumeClaim(claim)
-		if err == nil {
-			nextPhase = api.ClaimBound
-			glog.V(5).Infof("PersistentVolumeClaim[%s] is bound\n", claim.Name)
-		} else {
-			// Rollback by unsetting the ClaimRef on the volume pointer.
-			// the volume in the index will be unbound again and ready to be matched.
-			claim.Spec.VolumeName = ""
-			// Rollback by restoring original phase to claim pointer
-			nextPhase = api.ClaimPending
-			return fmt.Errorf("Error updating volume: %+v\n", err)
+		// create a reference to the claim and assign it to the volume being bound.
+		// the volume is a pointer and assigning the reference fixes a race condition where another
+		// claim might match this volume but before the claimRef is persistent in the next case statement
+		claimRef, err := api.GetReference(claim)
+		if err != nil {
+			return fmt.Errorf("Unexpected error getting claim reference: %v\n", err)
 		}
 
-	case api.ClaimBound:
-		volume, err := binderClient.GetPersistentVolume(claim.Spec.VolumeName)
+		// make a binding reference to the claim and ensure to update the local index to prevent dupe bindings
+		clone, err := conversion.NewCloner().DeepCopy(volume)
 		if err != nil {
-			return fmt.Errorf("Unexpected error getting persistent volume: %v\n", err)
+			return fmt.Errorf("Error cloning pv: %v", err)
+		}
+		volumeClone, ok := clone.(*api.PersistentVolume)
+		if !ok {
+			return fmt.Errorf("Unexpected pv cast error : %v\n", volumeClone)
+		}
+		volumeClone.Spec.ClaimRef = claimRef
+		if updatedVolume, err := binderClient.UpdatePersistentVolume(volumeClone); err != nil {
+			return fmt.Errorf("Unexpected error saving PersistentVolume.Status: %+v", err)
+		} else {
+			volume = updatedVolume
+			volumeIndex.Update(updatedVolume)
 		}
 
-		if volume.Spec.ClaimRef == nil {
-			glog.V(5).Infof("Rebuilding bind on pv.Spec.ClaimRef\n")
-			claimRef, err := api.GetReference(claim)
+		// the bind is persisted on the volume above and will always match the claim in a search.
+		// claim would remain Pending if the update fails, so processing this state is idempotent.
+		// this only needs to be processed once.
+		if claim.Spec.VolumeName != volume.Name {
+			claim.Spec.VolumeName = volume.Name
+			claim, err = binderClient.UpdatePersistentVolumeClaim(claim)
 			if err != nil {
-				return fmt.Errorf("Unexpected error getting claim reference: %v\n", err)
-			}
-			volume.Spec.ClaimRef = claimRef
-			_, err = binderClient.UpdatePersistentVolume(volume)
-			if err != nil {
-				return fmt.Errorf("Unexpected error saving PersistentVolume.Status: %+v", err)
+				return fmt.Errorf("Error updating claim with VolumeName %s: %+v\n", volume.Name, err)
 			}
 		}
 
-		// all "actuals" are transferred from PV to PVC so the user knows what
-		// type of volume they actually got for their claim.
-		// Volumes cannot have zero AccessModes, so checking that a claim has access modes
-		// is sufficient to tell us if these values have already been set.
-		if len(claim.Status.AccessModes) == 0 {
-			claim.Status.Phase = api.ClaimBound
-			claim.Status.AccessModes = volume.Spec.AccessModes
-			claim.Status.Capacity = volume.Spec.Capacity
-			_, err := binderClient.UpdatePersistentVolumeClaimStatus(claim)
-			if err != nil {
-				return fmt.Errorf("Unexpected error saving claim status: %+v", err)
-			}
+		claim.Status.Phase = api.ClaimBound
+		claim.Status.AccessModes = volume.Spec.AccessModes
+		claim.Status.Capacity = volume.Spec.Capacity
+		_, err = binderClient.UpdatePersistentVolumeClaimStatus(claim)
+		if err != nil {
+			return fmt.Errorf("Unexpected error saving claim status: %+v", err)
 		}
-	}
 
-	if currentPhase != nextPhase {
-		claim.Status.Phase = nextPhase
-		binderClient.UpdatePersistentVolumeClaimStatus(claim)
+	case api.ClaimBound:
+		// no-op.  Claim is bound, values from PV are set.  PVCs are technically mutable in the API server
+		// and we don't want to handle those changes at this time.
+
+	default:
+		return fmt.Errorf("Unknown state for PVC: %#v", claim)
+
 	}
+
+	glog.V(5).Infof("PersistentVolumeClaim[%s] is bound\n", claim.Name)
 	return nil
 }
 

pkg/controller/persistentvolume/persistentvolume_index_test.go

@@ -21,6 +21,7 @@ import (
 
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/resource"
+	"k8s.io/kubernetes/pkg/api/testapi"
 )
 
 func TestMatchVolume(t *testing.T) {
@@ -104,17 +105,17 @@ func TestMatchVolume(t *testing.T) {
 	}
 
 	for name, scenario := range scenarios {
-		volume, err := volList.FindBestMatchForClaim(scenario.claim)
+		volume, err := volList.findBestMatchForClaim(scenario.claim)
 		if err != nil {
 			t.Errorf("Unexpected error matching volume by claim: %v", err)
 		}
-		if scenario.expectedMatch != "" && volume == nil {
+		if len(scenario.expectedMatch) != 0 && volume == nil {
 			t.Errorf("Expected match but received nil volume for scenario: %s", name)
 		}
-		if scenario.expectedMatch != "" && volume != nil && string(volume.UID) != scenario.expectedMatch {
+		if len(scenario.expectedMatch) != 0 && volume != nil && string(volume.UID) != scenario.expectedMatch {
 			t.Errorf("Expected %s but got volume %s in scenario %s", scenario.expectedMatch, volume.UID, name)
 		}
-		if scenario.expectedMatch == "" && volume != nil {
+		if len(scenario.expectedMatch) == 0 && volume != nil {
 			t.Errorf("Unexpected match for scenario: %s", name)
 		}
 	}
@@ -175,7 +176,7 @@ func TestMatchingWithBoundVolumes(t *testing.T) {
 		},
 	}
 
-	volume, err := volumeIndex.FindBestMatchForClaim(claim)
+	volume, err := volumeIndex.findBestMatchForClaim(claim)
 	if err != nil {
 		t.Fatalf("Unexpected error matching volume by claim: %v", err)
 	}
@@ -296,27 +297,27 @@ func TestFindingVolumeWithDifferentAccessModes(t *testing.T) {
 	index.Add(ebs)
 	index.Add(nfs)
 
-	volume, _ := index.FindBestMatchForClaim(claim)
+	volume, _ := index.findBestMatchForClaim(claim)
 	if volume.Name != ebs.Name {
 		t.Errorf("Expected %s but got volume %s instead", ebs.Name, volume.Name)
 	}
 
 	claim.Spec.AccessModes = []api.PersistentVolumeAccessMode{api.ReadWriteOnce, api.ReadOnlyMany}
-	volume, _ = index.FindBestMatchForClaim(claim)
+	volume, _ = index.findBestMatchForClaim(claim)
 	if volume.Name != gce.Name {
 		t.Errorf("Expected %s but got volume %s instead", gce.Name, volume.Name)
 	}
 
 	// order of the requested modes should not matter
 	claim.Spec.AccessModes = []api.PersistentVolumeAccessMode{api.ReadWriteMany, api.ReadWriteOnce, api.ReadOnlyMany}
-	volume, _ = index.FindBestMatchForClaim(claim)
+	volume, _ = index.findBestMatchForClaim(claim)
 	if volume.Name != nfs.Name {
 		t.Errorf("Expected %s but got volume %s instead", nfs.Name, volume.Name)
 	}
 
 	// fewer modes requested should still match
 	claim.Spec.AccessModes = []api.PersistentVolumeAccessMode{api.ReadWriteMany}
-	volume, _ = index.FindBestMatchForClaim(claim)
+	volume, _ = index.findBestMatchForClaim(claim)
 	if volume.Name != nfs.Name {
 		t.Errorf("Expected %s but got volume %s instead", nfs.Name, volume.Name)
 	}
@@ -324,7 +325,7 @@ func TestFindingVolumeWithDifferentAccessModes(t *testing.T) {
 	// pretend the exact match is bound.  should get the next level up of modes.
 	ebs.Spec.ClaimRef = &api.ObjectReference{}
 	claim.Spec.AccessModes = []api.PersistentVolumeAccessMode{api.ReadWriteOnce}
-	volume, _ = index.FindBestMatchForClaim(claim)
+	volume, _ = index.findBestMatchForClaim(claim)
 	if volume.Name != gce.Name {
 		t.Errorf("Expected %s but got volume %s instead", gce.Name, volume.Name)
 	}
@@ -332,7 +333,7 @@ func TestFindingVolumeWithDifferentAccessModes(t *testing.T) {
 	// continue up the levels of modes.
 	gce.Spec.ClaimRef = &api.ObjectReference{}
 	claim.Spec.AccessModes = []api.PersistentVolumeAccessMode{api.ReadWriteOnce}
-	volume, _ = index.FindBestMatchForClaim(claim)
+	volume, _ = index.findBestMatchForClaim(claim)
 	if volume.Name != nfs.Name {
 		t.Errorf("Expected %s but got volume %s instead", nfs.Name, volume.Name)
 	}
@@ -340,7 +341,7 @@ func TestFindingVolumeWithDifferentAccessModes(t *testing.T) {
 	// partial mode request
 	gce.Spec.ClaimRef = nil
 	claim.Spec.AccessModes = []api.PersistentVolumeAccessMode{api.ReadOnlyMany}
-	volume, _ = index.FindBestMatchForClaim(claim)
+	volume, _ = index.findBestMatchForClaim(claim)
 	if volume.Name != gce.Name {
 		t.Errorf("Expected %s but got volume %s instead", gce.Name, volume.Name)
 	}
@@ -485,53 +486,40 @@ func createTestVolumes() []*api.PersistentVolume {
 	}
 }
 
-func TestFindingPreboundVolumes(t *testing.T) {
-	pv1 := &api.PersistentVolume{
+func testVolume(name, size string) *api.PersistentVolume {
+	return &api.PersistentVolume{
 		ObjectMeta: api.ObjectMeta{
-			Name:        "pv1",
+			Name:        name,
 			Annotations: map[string]string{},
 		},
 		Spec: api.PersistentVolumeSpec{
-			Capacity:               api.ResourceList{api.ResourceName(api.ResourceStorage): resource.MustParse("1Gi")},
-			PersistentVolumeSource: api.PersistentVolumeSource{HostPath: &api.HostPathVolumeSource{}},
-			AccessModes:            []api.PersistentVolumeAccessMode{api.ReadWriteOnce},
-		},
-	}
-
-	pv5 := &api.PersistentVolume{
-		ObjectMeta: api.ObjectMeta{
-			Name:        "pv5",
-			Annotations: map[string]string{},
-		},
-		Spec: api.PersistentVolumeSpec{
-			Capacity:               api.ResourceList{api.ResourceName(api.ResourceStorage): resource.MustParse("5Gi")},
-			PersistentVolumeSource: api.PersistentVolumeSource{HostPath: &api.HostPathVolumeSource{}},
-			AccessModes:            []api.PersistentVolumeAccessMode{api.ReadWriteOnce},
-		},
-	}
-
-	pv8 := &api.PersistentVolume{
-		ObjectMeta: api.ObjectMeta{
-			Name:        "pv8",
-			Annotations: map[string]string{},
-		},
-		Spec: api.PersistentVolumeSpec{
-			Capacity:               api.ResourceList{api.ResourceName(api.ResourceStorage): resource.MustParse("8Gi")},
+			Capacity:               api.ResourceList{api.ResourceName(api.ResourceStorage): resource.MustParse(size)},
 			PersistentVolumeSource: api.PersistentVolumeSource{HostPath: &api.HostPathVolumeSource{}},
 			AccessModes:            []api.PersistentVolumeAccessMode{api.ReadWriteOnce},
 		},
 	}
+}
 
+func TestFindingPreboundVolumes(t *testing.T) {
 	claim := &api.PersistentVolumeClaim{
 		ObjectMeta: api.ObjectMeta{
 			Name:      "claim01",
 			Namespace: "myns",
+			SelfLink:  testapi.Default.SelfLink("pvc", ""),
 		},
 		Spec: api.PersistentVolumeClaimSpec{
 			AccessModes: []api.PersistentVolumeAccessMode{api.ReadWriteOnce},
 			Resources:   api.ResourceRequirements{Requests: api.ResourceList{api.ResourceName(api.ResourceStorage): resource.MustParse("1Gi")}},
 		},
 	}
+	claimRef, err := api.GetReference(claim)
+	if err != nil {
+		t.Errorf("error getting claimRef: %v", err)
+	}
+
+	pv1 := testVolume("pv1", "1Gi")
+	pv5 := testVolume("pv5", "5Gi")
+	pv8 := testVolume("pv8", "8Gi")
 
 	index := NewPersistentVolumeOrderedIndex()
 	index.Add(pv1)
@@ -539,22 +527,22 @@ func TestFindingPreboundVolumes(t *testing.T) {
 	index.Add(pv8)
 
 	// expected exact match on size
-	volume, _ := index.FindBestMatchForClaim(claim)
+	volume, _ := index.findBestMatchForClaim(claim)
 	if volume.Name != pv1.Name {
 		t.Errorf("Expected %s but got volume %s instead", pv1.Name, volume.Name)
 	}
 
 	// pretend the exact match is pre-bound.  should get the next size up.
-	pv1.Annotations[createdForKey] = "some/other/claim"
-	volume, _ = index.FindBestMatchForClaim(claim)
+	pv1.Spec.ClaimRef = &api.ObjectReference{Name: "foo", Namespace: "bar"}
+	volume, _ = index.findBestMatchForClaim(claim)
 	if volume.Name != pv5.Name {
 		t.Errorf("Expected %s but got volume %s instead", pv5.Name, volume.Name)
 	}
 
 	// pretend the exact match is available but the largest volume is pre-bound to the claim.
-	delete(pv1.Annotations, createdForKey)
-	pv8.Annotations[createdForKey] = "myns/claim01"
-	volume, _ = index.FindBestMatchForClaim(claim)
+	pv1.Spec.ClaimRef = nil
+	pv8.Spec.ClaimRef = claimRef
+	volume, _ = index.findBestMatchForClaim(claim)
 	if volume.Name != pv8.Name {
 		t.Errorf("Expected %s but got volume %s instead", pv8.Name, volume.Name)
 	}

pkg/controller/persistentvolume/types.go

@@ -79,8 +79,8 @@ func (pvIndex *persistentVolumeOrderedIndex) ListByAccessModes(modes []api.Persi
 // matchPredicate is a function that indicates that a persistent volume matches another
 type matchPredicate func(compareThis, toThis *api.PersistentVolume) bool
 
-// Find returns the nearest PV from the ordered list or nil if a match is not found
-func (pvIndex *persistentVolumeOrderedIndex) Find(searchPV *api.PersistentVolume, matchPredicate matchPredicate) (*api.PersistentVolume, error) {
+// find returns the nearest PV from the ordered list or nil if a match is not found
+func (pvIndex *persistentVolumeOrderedIndex) find(searchPV *api.PersistentVolume, matchPredicate matchPredicate) (*api.PersistentVolume, error) {
 	// the 'searchPV' argument is a synthetic PV with capacity and accessmodes set according to the user's PersistentVolumeClaim.
 	// the synthetic pv arg is, therefore, a request for a storage resource.
 	//
@@ -94,6 +94,16 @@ func (pvIndex *persistentVolumeOrderedIndex) Find(searchPV *api.PersistentVolume
 	// potential matches (the GCEPD example above).
 	allPossibleModes := pvIndex.allPossibleMatchingAccessModes(searchPV.Spec.AccessModes)
 
+	// the searchPV should contain an annotation that allows pre-binding to a claim.
+	// we can use the same annotation value (pvc's namespace/name) and check against
+	// existing volumes to find an exact match. It is possible that a bind is made (ClaimRef persisted to PV)
+	// but the fail to update claim.Spec.VolumeName fails.  This check allows the claim to find the volume
+	// that's already bound to the claim.
+	preboundClaim := ""
+	if createdFor, ok := searchPV.Annotations[createdForKey]; ok {
+		preboundClaim = createdFor
+	}
+
 	for _, modes := range allPossibleModes {
 		volumes, err := pvIndex.ListByAccessModes(modes)
 		if err != nil {
@@ -105,23 +115,17 @@ func (pvIndex *persistentVolumeOrderedIndex) Find(searchPV *api.PersistentVolume
 		// return the exact pre-binding match, if found
 		unboundVolumes := []*api.PersistentVolume{}
 		for _, volume := range volumes {
-			// check for current binding
-			if volume.Spec.ClaimRef != nil {
+			if volume.Spec.ClaimRef == nil {
+				// volume isn't currently bound or pre-bound.
+				unboundVolumes = append(unboundVolumes, volume)
 				continue
 			}
 
-			// check for pre-bind where the volume is intended for one specific claim
-			if createdFor, ok := volume.Annotations[createdForKey]; ok {
-				if createdFor != searchPV.Annotations[createdForKey] {
-					// the volume is pre-bound and does not match the search criteria.
-					continue
-				}
-				// exact annotation match! No search required.
+			boundClaim := fmt.Sprintf("%s/%s", volume.Spec.ClaimRef.Namespace, volume.Spec.ClaimRef.Name)
+			if boundClaim == preboundClaim {
+				// exact match! No search required.
 				return volume, nil
 			}
-
-			// volume isn't currently bound or pre-bound.
-			unboundVolumes = append(unboundVolumes, volume)
 		}
 
 		i := sort.Search(len(unboundVolumes), func(i int) bool { return matchPredicate(searchPV, unboundVolumes[i]) })
@@ -147,11 +151,11 @@ func (pvIndex *persistentVolumeOrderedIndex) findByAccessModesAndStorageCapacity
 			},
 		},
 	}
-	return pvIndex.Find(pv, matchStorageCapacity)
+	return pvIndex.find(pv, matchStorageCapacity)
 }
 
-// FindBestMatchForClaim is a convenience method that finds a volume by the claim's AccessModes and requests for Storage
-func (pvIndex *persistentVolumeOrderedIndex) FindBestMatchForClaim(claim *api.PersistentVolumeClaim) (*api.PersistentVolume, error) {
+// findBestMatchForClaim is a convenience method that finds a volume by the claim's AccessModes and requests for Storage
+func (pvIndex *persistentVolumeOrderedIndex) findBestMatchForClaim(claim *api.PersistentVolumeClaim) (*api.PersistentVolume, error) {
 	return pvIndex.findByAccessModesAndStorageCapacity(fmt.Sprintf("%s/%s", claim.Namespace, claim.Name), claim.Spec.AccessModes, claim.Spec.Resources.Requests[api.ResourceName(api.ResourceStorage)])
 }
 

test/integration/persistent_volumes_test.go

@@ -19,6 +19,8 @@ limitations under the License.
 package integration
 
 import (
+	"fmt"
+	"math/rand"
 	"testing"
 	"time"
 
@@ -28,6 +30,7 @@ import (
 	"k8s.io/kubernetes/pkg/api/unversioned"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
 	persistentvolumecontroller "k8s.io/kubernetes/pkg/controller/persistentvolume"
+	"k8s.io/kubernetes/pkg/conversion"
 	"k8s.io/kubernetes/pkg/fields"
 	"k8s.io/kubernetes/pkg/labels"
 	"k8s.io/kubernetes/pkg/volume"
@@ -48,11 +51,11 @@ func TestPersistentVolumeRecycler(t *testing.T) {
 	recyclerClient := client.NewOrDie(&client.Config{Host: s.URL, GroupVersion: testapi.Default.GroupVersion()})
 	testClient := client.NewOrDie(&client.Config{Host: s.URL, GroupVersion: testapi.Default.GroupVersion()})
 
-	binder := persistentvolumecontroller.NewPersistentVolumeClaimBinder(binderClient, 1*time.Second)
+	binder := persistentvolumecontroller.NewPersistentVolumeClaimBinder(binderClient, 10*time.Minute)
 	binder.Run()
 	defer binder.Stop()
 
-	recycler, _ := persistentvolumecontroller.NewPersistentVolumeRecycler(recyclerClient, 1*time.Second, []volume.VolumePlugin{&volume.FakeVolumePlugin{"plugin-name", volume.NewFakeVolumeHost("/tmp/fake", nil, nil)}})
+	recycler, _ := persistentvolumecontroller.NewPersistentVolumeRecycler(recyclerClient, 30*time.Minute, []volume.VolumePlugin{&volume.FakeVolumePlugin{"plugin-name", volume.NewFakeVolumeHost("/tmp/fake", nil, nil)}})
 	recycler.Run()
 	defer recycler.Stop()
 
@@ -122,6 +125,50 @@ func TestPersistentVolumeRecycler(t *testing.T) {
 			break
 		}
 	}
+
+	// test the race between claims and volumes.  ensure only a volume only binds to a single claim.
+	deleteAllEtcdKeys()
+	counter := 0
+	maxClaims := 100
+	claims := []*api.PersistentVolumeClaim{}
+	for counter <= maxClaims {
+		counter += 1
+		clone, _ := conversion.NewCloner().DeepCopy(pvc)
+		newPvc, _ := clone.(*api.PersistentVolumeClaim)
+		newPvc.ObjectMeta = api.ObjectMeta{Name: fmt.Sprintf("fake-pvc-%d", counter)}
+		claim, err := testClient.PersistentVolumeClaims(api.NamespaceDefault).Create(newPvc)
+		if err != nil {
+			t.Fatal("Error creating newPvc: %v", err)
+		}
+		claims = append(claims, claim)
+	}
+
+	// putting a bind manually on a pv should only match the claim it is bound to
+	rand.Seed(time.Now().Unix())
+	claim := claims[rand.Intn(maxClaims-1)]
+	claimRef, err := api.GetReference(claim)
+	if err != nil {
+		t.Fatalf("Unexpected error getting claimRef: %v", err)
+	}
+	pv.Spec.ClaimRef = claimRef
+
+	pv, err = testClient.PersistentVolumes().Create(pv)
+	if err != nil {
+		t.Fatalf("Unexpected error creating pv: %v", err)
+	}
+
+	waitForPersistentVolumePhase(w, api.VolumeBound)
+
+	pv, err = testClient.PersistentVolumes().Get(pv.Name)
+	if err != nil {
+		t.Fatalf("Unexpected error getting pv: %v", err)
+	}
+	if pv.Spec.ClaimRef == nil {
+		t.Fatalf("Unexpected nil claimRef")
+	}
+	if pv.Spec.ClaimRef.Namespace != claimRef.Namespace || pv.Spec.ClaimRef.Name != claimRef.Name {
+		t.Fatalf("Bind mismatch! Expected %s/%s but got %s/%s", claimRef.Namespace, claimRef.Name, pv.Spec.ClaimRef.Namespace, pv.Spec.ClaimRef.Name)
+	}
 }
 
 func waitForPersistentVolumePhase(w watch.Interface, phase api.PersistentVolumePhase) {