Merge pull request kubernetes#10833 from markturansky/loosen_binding

Allow PV/Claim matching on inexact 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"
@@ -178,3 +179,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,23 +74,38 @@ 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)
- if err != nil {
- return nil, err
- }
+ // 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
+ }
 
- // 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)
+ // 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
+ 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
 }
@@ -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)
+}