added better matching for PV access modes

pkg/api/helpers.go

@@ -20,6 +20,7 @@ import (
 	"crypto/md5"
 	"fmt"
 	"reflect"
+	"strings"
 
 	"k8s.io/kubernetes/pkg/api/resource"
 	"k8s.io/kubernetes/pkg/conversion"
@@ -179,3 +180,58 @@ func LoadBalancerStatusDeepCopy(lb *LoadBalancerStatus) *LoadBalancerStatus {
 	}
 	return c
 }
+
+// GetAccessModesAsString returns a string representation of an array of access modes.
+// modes, when present, are always in the same order: RWO,ROX,RWX.
+func GetAccessModesAsString(modes []PersistentVolumeAccessMode) string {
+	modes = removeDuplicateAccessModes(modes)
+	modesStr := []string{}
+	if containsAccessMode(modes, ReadWriteOnce) {
+		modesStr = append(modesStr, "RWO")
+	}
+	if containsAccessMode(modes, ReadOnlyMany) {
+		modesStr = append(modesStr, "ROX")
+	}
+	if containsAccessMode(modes, ReadWriteMany) {
+		modesStr = append(modesStr, "RWX")
+	}
+	return strings.Join(modesStr, ",")
+}
+
+// GetAccessModesAsString returns an array of AccessModes from a string created by GetAccessModesAsString
+func GetAccessModesFromString(modes string) []PersistentVolumeAccessMode {
+	strmodes := strings.Split(modes, ",")
+	accessModes := []PersistentVolumeAccessMode{}
+	for _, s := range strmodes {
+		s = strings.Trim(s, " ")
+		switch {
+		case s == "RWO":
+			accessModes = append(accessModes, ReadWriteOnce)
+		case s == "ROX":
+			accessModes = append(accessModes, ReadOnlyMany)
+		case s == "RWX":
+			accessModes = append(accessModes, ReadWriteMany)
+		}
+	}
+	return accessModes
+}
+
+// removeDuplicateAccessModes returns an array of access modes without any duplicates
+func removeDuplicateAccessModes(modes []PersistentVolumeAccessMode) []PersistentVolumeAccessMode {
+	accessModes := []PersistentVolumeAccessMode{}
+	for _, m := range modes {
+		if !containsAccessMode(accessModes, m) {
+			accessModes = append(accessModes, m)
+		}
+	}
+	return accessModes
+}
+
+func containsAccessMode(modes []PersistentVolumeAccessMode, mode PersistentVolumeAccessMode) bool {
+	for _, m := range modes {
+		if m == mode {
+			return true
+		}
+	}
+	return false
+}

pkg/api/helpers_test.go

@@ -142,3 +142,36 @@ func TestAddToNodeAddresses(t *testing.T) {
 		}
 	}
 }
+
+func TestGetAccessModesFromString(t *testing.T) {
+	modes := GetAccessModesFromString("ROX")
+	if !containsAccessMode(modes, ReadOnlyMany) {
+		t.Errorf("Expected mode %s, but got %+v", ReadOnlyMany, modes)
+	}
+
+	modes = GetAccessModesFromString("ROX,RWX")
+	if !containsAccessMode(modes, ReadOnlyMany) {
+		t.Errorf("Expected mode %s, but got %+v", ReadOnlyMany, modes)
+	}
+	if !containsAccessMode(modes, ReadWriteMany) {
+		t.Errorf("Expected mode %s, but got %+v", ReadWriteMany, modes)
+	}
+
+	modes = GetAccessModesFromString("RWO,ROX,RWX")
+	if !containsAccessMode(modes, ReadOnlyMany) {
+		t.Errorf("Expected mode %s, but got %+v", ReadOnlyMany, modes)
+	}
+	if !containsAccessMode(modes, ReadWriteMany) {
+		t.Errorf("Expected mode %s, but got %+v", ReadWriteMany, modes)
+	}
+}
+
+func TestRemoveDuplicateAccessModes(t *testing.T) {
+	modes := []PersistentVolumeAccessMode{
+		ReadWriteOnce, ReadOnlyMany, ReadOnlyMany, ReadOnlyMany,
+	}
+	modes = removeDuplicateAccessModes(modes)
+	if len(modes) != 2 {
+		t.Errorf("Expected 2 distinct modes in set but found %v", len(modes))
+	}
+}

pkg/controller/persistentvolume/persistentvolume_index_test.go

@@ -216,6 +216,136 @@ func TestSort(t *testing.T) {
 	}
 }
 
+func TestAllPossibleAccessModes(t *testing.T) {
+	index := NewPersistentVolumeOrderedIndex()
+	for _, pv := range createTestVolumes() {
+		index.Add(pv)
+	}
+
+	// the mock PVs creates contain 2 types of accessmodes:   RWO+ROX and RWO+ROW+RWX
+	possibleModes := index.allPossibleMatchingAccessModes([]api.PersistentVolumeAccessMode{api.ReadWriteOnce})
+	if len(possibleModes) != 2 {
+		t.Errorf("Expected 2 arrays of modes that match RWO, but got %v", len(possibleModes))
+	}
+	for _, m := range possibleModes {
+		if !contains(m, api.ReadWriteOnce) {
+			t.Errorf("AccessModes does not contain %s", api.ReadWriteOnce)
+		}
+	}
+
+	possibleModes = index.allPossibleMatchingAccessModes([]api.PersistentVolumeAccessMode{api.ReadWriteMany})
+	if len(possibleModes) != 1 {
+		t.Errorf("Expected 1 array of modes that match RWX, but got %v", len(possibleModes))
+	}
+	if !contains(possibleModes[0], api.ReadWriteMany) {
+		t.Errorf("AccessModes does not contain %s", api.ReadWriteOnce)
+	}
+
+}
+
+func TestFindingVolumeWithDifferentAccessModes(t *testing.T) {
+	gce := &api.PersistentVolume{
+		ObjectMeta: api.ObjectMeta{UID: "001", Name: "gce"},
+		Spec: api.PersistentVolumeSpec{
+			Capacity:               api.ResourceList{api.ResourceName(api.ResourceStorage): resource.MustParse("10G")},
+			PersistentVolumeSource: api.PersistentVolumeSource{GCEPersistentDisk: &api.GCEPersistentDiskVolumeSource{}},
+			AccessModes: []api.PersistentVolumeAccessMode{
+				api.ReadWriteOnce,
+				api.ReadOnlyMany,
+			},
+		},
+	}
+
+	ebs := &api.PersistentVolume{
+		ObjectMeta: api.ObjectMeta{UID: "002", Name: "ebs"},
+		Spec: api.PersistentVolumeSpec{
+			Capacity:               api.ResourceList{api.ResourceName(api.ResourceStorage): resource.MustParse("10G")},
+			PersistentVolumeSource: api.PersistentVolumeSource{AWSElasticBlockStore: &api.AWSElasticBlockStoreVolumeSource{}},
+			AccessModes: []api.PersistentVolumeAccessMode{
+				api.ReadWriteOnce,
+			},
+		},
+	}
+
+	nfs := &api.PersistentVolume{
+		ObjectMeta: api.ObjectMeta{UID: "003", Name: "nfs"},
+		Spec: api.PersistentVolumeSpec{
+			Capacity:               api.ResourceList{api.ResourceName(api.ResourceStorage): resource.MustParse("10G")},
+			PersistentVolumeSource: api.PersistentVolumeSource{NFS: &api.NFSVolumeSource{}},
+			AccessModes: []api.PersistentVolumeAccessMode{
+				api.ReadWriteOnce,
+				api.ReadOnlyMany,
+				api.ReadWriteMany,
+			},
+		},
+	}
+
+	claim := &api.PersistentVolumeClaim{
+		ObjectMeta: api.ObjectMeta{
+			Name:      "claim01",
+			Namespace: "myns",
+		},
+		Spec: api.PersistentVolumeClaimSpec{
+			AccessModes: []api.PersistentVolumeAccessMode{api.ReadWriteOnce},
+			Resources:   api.ResourceRequirements{Requests: api.ResourceList{api.ResourceName(api.ResourceStorage): resource.MustParse("1G")}},
+		},
+	}
+
+	index := NewPersistentVolumeOrderedIndex()
+	index.Add(gce)
+	index.Add(ebs)
+	index.Add(nfs)
+
+	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)
+	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)
+	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)
+	if volume.Name != nfs.Name {
+		t.Errorf("Expected %s but got volume %s instead", nfs.Name, volume.Name)
+	}
+
+	// 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)
+	if volume.Name != gce.Name {
+		t.Errorf("Expected %s but got volume %s instead", gce.Name, volume.Name)
+	}
+
+	// continue up the levels of modes.
+	gce.Spec.ClaimRef = &api.ObjectReference{}
+	claim.Spec.AccessModes = []api.PersistentVolumeAccessMode{api.ReadWriteOnce}
+	volume, _ = index.FindBestMatchForClaim(claim)
+	if volume.Name != nfs.Name {
+		t.Errorf("Expected %s but got volume %s instead", nfs.Name, volume.Name)
+	}
+
+	// partial mode request
+	gce.Spec.ClaimRef = nil
+	claim.Spec.AccessModes = []api.PersistentVolumeAccessMode{api.ReadOnlyMany}
+	volume, _ = index.FindBestMatchForClaim(claim)
+	if volume.Name != gce.Name {
+		t.Errorf("Expected %s but got volume %s instead", gce.Name, volume.Name)
+	}
+}
+
 func createTestVolumes() []*api.PersistentVolume {
 	// these volumes are deliberately out-of-order to test indexing and sorting
 	return []*api.PersistentVolume{

pkg/controller/persistentvolume/types.go

@@ -23,7 +23,6 @@ import (
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/resource"
 	"k8s.io/kubernetes/pkg/client/unversioned/cache"
-	"k8s.io/kubernetes/pkg/volume"
 )
 
 // persistentVolumeOrderedIndex is a cache.Store that keeps persistent volumes indexed by AccessModes and ordered by storage capacity.
@@ -42,7 +41,7 @@ func NewPersistentVolumeOrderedIndex() *persistentVolumeOrderedIndex {
 // accessModesIndexFunc is an indexing function that returns a persistent volume's AccessModes as a string
 func accessModesIndexFunc(obj interface{}) ([]string, error) {
 	if pv, ok := obj.(*api.PersistentVolume); ok {
-		modes := volume.GetAccessModesAsString(pv.Spec.AccessModes)
+		modes := api.GetAccessModesAsString(pv.Spec.AccessModes)
 		return []string{modes}, nil
 	}
 	return []string{""}, fmt.Errorf("object is not a persistent volume: %v", obj)
@@ -75,7 +74,21 @@ 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(pv *api.PersistentVolume, matchPredicate matchPredicate) (*api.PersistentVolume, error) {
-	volumes, err := pvIndex.ListByAccessModes(pv.Spec.AccessModes)
+	// the 'pv' 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.
+	//
+	// PVs are indexed by their access modes to allow easier searching.  Each index is the string representation of a set of access modes.
+	// There is a finite number of possible sets and PVs will only be indexed in one of them (whichever index matches the PV's modes).
+	//
+	// A request for resources will always specify its desired access modes.  Any matching PV must have at least that number
+	// of access modes, but it can have more.  For example, a user asks for ReadWriteOnce but a GCEPD is available, which is ReadWriteOnce+ReadOnlyMany.
+	//
+	// Searches are performed against a set of access modes, so we can attempt not only the exact matching modes but also
+	// potential matches (the GCEPD example above).
+	allPossibleModes := pvIndex.allPossibleMatchingAccessModes(pv.Spec.AccessModes)
+
+	for _, modes := range allPossibleModes {
+		volumes, err := pvIndex.ListByAccessModes(modes)
 		if err != nil {
 			return nil, err
 		}
@@ -93,6 +106,7 @@ func (pvIndex *persistentVolumeOrderedIndex) Find(pv *api.PersistentVolume, matc
 		if i < len(unboundVolumes) {
 			return unboundVolumes[i], nil
 		}
+	}
 	return nil, nil
 }
 
@@ -139,3 +153,87 @@ func matchStorageCapacity(pvA, pvB *api.PersistentVolume) bool {
 	bSize := bQty.Value()
 	return aSize <= bSize
 }
+
+// allPossibleMatchingAccessModes returns an array of AccessMode arrays that can satisfy a user's requested modes.
+//
+// see comments in the Find func above regarding indexing.
+//
+// allPossibleMatchingAccessModes gets all stringified accessmodes from the index and returns all those that
+// contain at least all of the requested mode.
+//
+// For example, assume the index contains 2 types of PVs where the stringified accessmodes are:
+//
+// "RWO,ROX" -- some number of GCEPDs
+// "RWO,ROX,RWX" -- some number of NFS volumes
+//
+// A request for RWO could be satisfied by both sets of indexed volumes, so allPossibleMatchingAccessModes returns:
+//
+// [][]api.PersistentVolumeAccessMode {
+//      []api.PersistentVolumeAccessMode {
+//			api.ReadWriteOnce, api.ReadOnlyMany,
+//		},
+//      []api.PersistentVolumeAccessMode {
+//			api.ReadWriteOnce, api.ReadOnlyMany, api.ReadWriteMany,
+//		},
+// }
+//
+// A request for RWX can be satisfied by only one set of indexed volumes, so the return is:
+//
+// [][]api.PersistentVolumeAccessMode {
+//      []api.PersistentVolumeAccessMode {
+//			api.ReadWriteOnce, api.ReadOnlyMany, api.ReadWriteMany,
+//		},
+// }
+//
+// This func returns modes with ascending levels of modes to give the user what is closest to what they actually asked for.
+//
+func (pvIndex *persistentVolumeOrderedIndex) allPossibleMatchingAccessModes(requestedModes []api.PersistentVolumeAccessMode) [][]api.PersistentVolumeAccessMode {
+	matchedModes := [][]api.PersistentVolumeAccessMode{}
+	keys := pvIndex.Indexer.ListIndexFuncValues("accessmodes")
+	for _, key := range keys {
+		indexedModes := api.GetAccessModesFromString(key)
+		if containedInAll(indexedModes, requestedModes) {
+			matchedModes = append(matchedModes, indexedModes)
+		}
+	}
+
+	// sort by the number of modes in each array with the fewest number of modes coming first.
+	// this allows searching for volumes by the minimum number of modes required of the possible matches.
+	sort.Sort(byAccessModes{matchedModes})
+	return matchedModes
+}
+
+func contains(modes []api.PersistentVolumeAccessMode, mode api.PersistentVolumeAccessMode) bool {
+	for _, m := range modes {
+		if m == mode {
+			return true
+		}
+	}
+	return false
+}
+
+func containedInAll(indexedModes []api.PersistentVolumeAccessMode, requestedModes []api.PersistentVolumeAccessMode) bool {
+	for _, mode := range requestedModes {
+		if !contains(indexedModes, mode) {
+			return false
+		}
+	}
+	return true
+}
+
+// byAccessModes is used to order access modes by size, with the fewest modes first
+type byAccessModes struct {
+	modes [][]api.PersistentVolumeAccessMode
+}
+
+func (c byAccessModes) Less(i, j int) bool {
+	return len(c.modes[i]) < len(c.modes[j])
+}
+
+func (c byAccessModes) Swap(i, j int) {
+	c.modes[i], c.modes[j] = c.modes[j], c.modes[i]
+}
+
+func (c byAccessModes) Len() int {
+	return len(c.modes)
+}

pkg/kubectl/describe.go

@@ -34,7 +34,6 @@ import (
 	"k8s.io/kubernetes/pkg/labels"
 	"k8s.io/kubernetes/pkg/types"
 	"k8s.io/kubernetes/pkg/util"
-	"k8s.io/kubernetes/pkg/volume"
 )
 
 // Describer generates output for the named resource or an error
@@ -599,7 +598,7 @@ func (d *PersistentVolumeDescriber) Describe(namespace, name string) (string, er
 			fmt.Fprintf(out, "Claim:\t%s\n", "")
 		}
 		fmt.Fprintf(out, "Reclaim Policy:\t%v\n", pv.Spec.PersistentVolumeReclaimPolicy)
-		fmt.Fprintf(out, "Access Modes:\t%s\n", volume.GetAccessModesAsString(pv.Spec.AccessModes))
+		fmt.Fprintf(out, "Access Modes:\t%s\n", api.GetAccessModesAsString(pv.Spec.AccessModes))
 		fmt.Fprintf(out, "Capacity:\t%s\n", storage.String())
 		fmt.Fprintf(out, "Message:\t%s\n", pv.Status.Message)
 		fmt.Fprintf(out, "Source:\n")
@@ -642,7 +641,7 @@ func (d *PersistentVolumeClaimDescriber) Describe(namespace, name string) (strin
 	capacity := ""
 	accessModes := ""
 	if pvc.Spec.VolumeName != "" {
-		accessModes = volume.GetAccessModesAsString(pvc.Status.AccessModes)
+		accessModes = api.GetAccessModesAsString(pvc.Status.AccessModes)
 		storage = pvc.Status.Capacity[api.ResourceStorage]
 		capacity = storage.String()
 	}

pkg/kubectl/resource_printer.go

@@ -40,7 +40,6 @@ import (
 	"k8s.io/kubernetes/pkg/runtime"
 	"k8s.io/kubernetes/pkg/util"
 	"k8s.io/kubernetes/pkg/util/jsonpath"
-	"k8s.io/kubernetes/pkg/volume"
 )
 
 // GetPrinter takes a format type, an optional format argument. It will return true
@@ -872,7 +871,7 @@ func printPersistentVolume(pv *api.PersistentVolume, w io.Writer, withNamespace
 		claimRefUID += pv.Spec.ClaimRef.Name
 	}
 
-	modesStr := volume.GetAccessModesAsString(pv.Spec.AccessModes)
+	modesStr := api.GetAccessModesAsString(pv.Spec.AccessModes)
 
 	aQty := pv.Spec.Capacity[api.ResourceStorage]
 	aSize := aQty.String()
@@ -917,7 +916,7 @@ func printPersistentVolumeClaim(pvc *api.PersistentVolumeClaim, w io.Writer, wit
 	capacity := ""
 	accessModes := ""
 	if pvc.Spec.VolumeName != "" {
-		accessModes = volume.GetAccessModesAsString(pvc.Status.AccessModes)
+		accessModes = api.GetAccessModesAsString(pvc.Status.AccessModes)
 		storage = pvc.Status.Capacity[api.ResourceStorage]
 		capacity = storage.String()
 	}

pkg/volume/util.go

@@ -31,38 +31,6 @@ import (
 	"github.com/golang/glog"
 )
 
-func GetAccessModesAsString(modes []api.PersistentVolumeAccessMode) string {
-	modesAsString := ""
-
-	if contains(modes, api.ReadWriteOnce) {
-		appendAccessMode(&modesAsString, "RWO")
-	}
-	if contains(modes, api.ReadOnlyMany) {
-		appendAccessMode(&modesAsString, "ROX")
-	}
-	if contains(modes, api.ReadWriteMany) {
-		appendAccessMode(&modesAsString, "RWX")
-	}
-
-	return modesAsString
-}
-
-func appendAccessMode(modes *string, mode string) {
-	if *modes != "" {
-		*modes += ","
-	}
-	*modes += mode
-}
-
-func contains(modes []api.PersistentVolumeAccessMode, mode api.PersistentVolumeAccessMode) bool {
-	for _, m := range modes {
-		if m == mode {
-			return true
-		}
-	}
-	return false
-}
-
 // ScrubPodVolumeAndWatchUntilCompletion is intended for use with volume Recyclers.  This function will
 // save the given Pod to the API and watch it until it completes, fails, or the pod's ActiveDeadlineSeconds is exceeded, whichever comes first.
 // An attempt to delete a scrubber pod is always attempted before returning.