fixed search bug by making an explicit pass to filter bound volumes

pkg/volumeclaimbinder/persistent_volume_index_test.go

@@ -112,7 +112,7 @@ func TestMatchVolume(t *testing.T) {
 			t.Errorf("Expected match but received nil volume for scenario: %s", name)
 		}
 		if scenario.expectedMatch != "" && volume != nil && string(volume.UID) != scenario.expectedMatch {
-			t.Errorf("Expected %s but got volume %s instead", scenario.expectedMatch, volume.UID)
+			t.Errorf("Expected %s but got volume %s in scenario %s", scenario.expectedMatch, volume.UID, name)
 		}
 		if scenario.expectedMatch == "" && volume != nil {
 			t.Errorf("Unexpected match for scenario: %s", name)
@@ -120,6 +120,73 @@ func TestMatchVolume(t *testing.T) {
 	}
 }
 
+func TestMatchingWithBoundVolumes(t *testing.T) {
+	volumeIndex := NewPersistentVolumeOrderedIndex()
+	// two similar volumes, one is bound
+	pv1 := &api.PersistentVolume{
+		ObjectMeta: api.ObjectMeta{
+			UID:  "gce-pd-1",
+			Name: "gce001",
+		},
+		Spec: api.PersistentVolumeSpec{
+			Capacity: api.ResourceList{
+				api.ResourceName(api.ResourceStorage): resource.MustParse("1G"),
+			},
+			PersistentVolumeSource: api.PersistentVolumeSource{
+				GCEPersistentDisk: &api.GCEPersistentDiskVolumeSource{},
+			},
+			AccessModes: []api.PersistentVolumeAccessMode{api.ReadWriteOnce, api.ReadOnlyMany},
+			// this one we're pretending is already bound
+			ClaimRef: &api.ObjectReference{UID: "abc123"},
+		},
+	}
+
+	pv2 := &api.PersistentVolume{
+		ObjectMeta: api.ObjectMeta{
+			UID:  "gce-pd-2",
+			Name: "gce002",
+		},
+		Spec: api.PersistentVolumeSpec{
+			Capacity: api.ResourceList{
+				api.ResourceName(api.ResourceStorage): resource.MustParse("1G"),
+			},
+			PersistentVolumeSource: api.PersistentVolumeSource{
+				GCEPersistentDisk: &api.GCEPersistentDiskVolumeSource{},
+			},
+			AccessModes: []api.PersistentVolumeAccessMode{api.ReadWriteOnce, api.ReadOnlyMany},
+		},
+	}
+
+	volumeIndex.Add(pv1)
+	volumeIndex.Add(pv2)
+
+	claim := &api.PersistentVolumeClaim{
+		ObjectMeta: api.ObjectMeta{
+			Name:      "claim01",
+			Namespace: "myns",
+		},
+		Spec: api.PersistentVolumeClaimSpec{
+			AccessModes: []api.PersistentVolumeAccessMode{api.ReadOnlyMany, api.ReadWriteOnce},
+			Resources: api.ResourceRequirements{
+				Requests: api.ResourceList{
+					api.ResourceName(api.ResourceStorage): resource.MustParse("1G"),
+				},
+			},
+		},
+	}
+
+	volume, err := volumeIndex.FindBestMatchForClaim(claim)
+	if err != nil {
+		t.Fatalf("Unexpected error matching volume by claim: %v", err)
+	}
+	if volume == nil {
+		t.Fatalf("Unexpected nil volume.  Expected %s", pv2.Name)
+	}
+	if pv2.Name != volume.Name {
+		t.Errorf("Expected %s but got volume %s instead", pv2.Name, volume.Name)
+	}
+}
+
 func TestSort(t *testing.T) {
 	volList := NewPersistentVolumeOrderedIndex()
 	for _, pv := range createTestVolumes() {
@@ -170,6 +237,26 @@ func createTestVolumes() []*api.PersistentVolume {
 				},
 			},
 		},
+		{
+			ObjectMeta: api.ObjectMeta{
+				UID:  "gce-pd-20",
+				Name: "gce004",
+			},
+			Spec: api.PersistentVolumeSpec{
+				Capacity: api.ResourceList{
+					api.ResourceName(api.ResourceStorage): resource.MustParse("20G"),
+				},
+				PersistentVolumeSource: api.PersistentVolumeSource{
+					GCEPersistentDisk: &api.GCEPersistentDiskVolumeSource{},
+				},
+				AccessModes: []api.PersistentVolumeAccessMode{
+					api.ReadWriteOnce,
+					api.ReadOnlyMany,
+				},
+				// this one we're pretending is already bound
+				ClaimRef: &api.ObjectReference{UID: "def456"},
+			},
+		},
 		{
 			ObjectMeta: api.ObjectMeta{
 				UID:  "nfs-5",

pkg/volumeclaimbinder/types.go

@@ -80,9 +80,18 @@ func (pvIndex *persistentVolumeOrderedIndex) Find(pv *api.PersistentVolume, matc
 		return nil, err
 	}
 
-	i := sort.Search(len(volumes), func(i int) bool { return matchPredicate(pv, volumes[i]) })
-	if i < len(volumes) {
-		return volumes[i], nil
+	// volumes are sorted by size but some may be bound.
+	// remove bound volumes for easy binary search by size
+	unboundVolumes := []*api.PersistentVolume{}
+	for _, v := range volumes {
+		if v.Spec.ClaimRef == nil {
+			unboundVolumes = append(unboundVolumes, v)
+		}
+	}
+
+	i := sort.Search(len(unboundVolumes), func(i int) bool { return matchPredicate(pv, unboundVolumes[i]) })
+	if i < len(unboundVolumes) {
+		return unboundVolumes[i], nil
 	}
 	return nil, nil
 }
@@ -97,8 +106,7 @@ func (pvIndex *persistentVolumeOrderedIndex) FindByAccessModesAndStorageCapacity
 			},
 		},
 	}
-
-	return pvIndex.Find(pv, filterBoundVolumes)
+	return pvIndex.Find(pv, matchStorageCapacity)
 }
 
 // FindBestMatchForClaim is a convenience method that finds a volume by the claim's AccessModes and requests for Storage
@@ -125,22 +133,9 @@ func (c byCapacity) Len() int {
 
 // matchStorageCapacity is a matchPredicate used to sort and find volumes
 func matchStorageCapacity(pvA, pvB *api.PersistentVolume) bool {
-	// skip already claimed volumes
-	if pvA.Spec.ClaimRef != nil {
-		return false
-	}
-
 	aQty := pvA.Spec.Capacity[api.ResourceStorage]
 	bQty := pvB.Spec.Capacity[api.ResourceStorage]
 	aSize := aQty.Value()
 	bSize := bQty.Value()
 	return aSize <= bSize
 }
-
-// filterBoundVolumes is a matchPredicate that filters bound volumes before comparing storage capacity
-func filterBoundVolumes(compareThis, toThis *api.PersistentVolume) bool {
-	if compareThis.Spec.ClaimRef != nil || toThis.Spec.ClaimRef != nil {
-		return false
-	}
-	return matchStorageCapacity(compareThis, toThis)
-}