Implement volume plugin changes for volume limits

pkg/volume/aws_ebs/aws_ebs.go

@@ -17,9 +17,11 @@ limitations under the License.
 package aws_ebs
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"path/filepath"
+	"regexp"
 	"strconv"
 	"strings"
 
@@ -95,6 +97,39 @@ func (plugin *awsElasticBlockStorePlugin) SupportsBulkVolumeVerification() bool
 	return true
 }
 
+func (plugin *awsElasticBlockStorePlugin) GetVolumeLimits() (map[string]int64, error) {
+	cloud := plugin.host.GetCloudProvider()
+
+	if cloud.ProviderName() != aws.ProviderName {
+		return nil, fmt.Errorf("Expected aws cloud, found %s", cloud.ProviderName())
+	}
+
+	volumeLimits := map[string]int64{
+		util.EBSVolumeLimitKey: 39,
+	}
+	instances, ok := cloud.Instances()
+	if !ok {
+		glog.V(3).Infof("Failed to get instances from cloud provider")
+		return volumeLimits, nil
+	}
+
+	instanceType, err := instances.InstanceType(context.TODO(), plugin.host.GetNodeName())
+	if err != nil {
+		glog.Errorf("Failed to get instance type from AWS cloud provider")
+		return volumeLimits, nil
+	}
+
+	if ok, _ := regexp.MatchString("^[cm]5.*", instanceType); ok {
+		volumeLimits[util.EBSVolumeLimitKey] = 25
+	}
+
+	return volumeLimits, nil
+}
+
+func (plugin *awsElasticBlockStorePlugin) VolumeLimitKey(spec *volume.Spec) string {
+	return util.EBSVolumeLimitKey
+}
+
 func (plugin *awsElasticBlockStorePlugin) GetAccessModes() []v1.PersistentVolumeAccessMode {
 	return []v1.PersistentVolumeAccessMode{
 		v1.ReadWriteOnce,
@@ -267,6 +302,7 @@ func (plugin *awsElasticBlockStorePlugin) ExpandVolumeDevice(
 }
 
 var _ volume.ExpandableVolumePlugin = &awsElasticBlockStorePlugin{}
+var _ volume.VolumePluginWithAttachLimits = &awsElasticBlockStorePlugin{}
 
 // Abstract interface to PD operations.
 type ebsManager interface {

pkg/volume/azure_dd/azure_dd.go

@@ -17,13 +17,17 @@ limitations under the License.
 package azure_dd
 
 import (
+	"fmt"
+
 	"github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2017-12-01/compute"
 	"github.com/Azure/azure-sdk-for-go/services/storage/mgmt/2017-10-01/storage"
 	"github.com/golang/glog"
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/kubernetes/pkg/cloudprovider/providers/azure"
 	"k8s.io/kubernetes/pkg/util/mount"
 	"k8s.io/kubernetes/pkg/volume"
+	"k8s.io/kubernetes/pkg/volume/util"
 )
 
 // interface exposed by the cloud provider implementing Disk functionality
@@ -62,6 +66,7 @@ var _ volume.PersistentVolumePlugin = &azureDataDiskPlugin{}
 var _ volume.DeletableVolumePlugin = &azureDataDiskPlugin{}
 var _ volume.ProvisionableVolumePlugin = &azureDataDiskPlugin{}
 var _ volume.AttachableVolumePlugin = &azureDataDiskPlugin{}
+var _ volume.VolumePluginWithAttachLimits = &azureDataDiskPlugin{}
 
 const (
 	azureDataDiskPluginName = "kubernetes.io/azure-disk"
@@ -106,6 +111,22 @@ func (plugin *azureDataDiskPlugin) SupportsBulkVolumeVerification() bool {
 	return false
 }
 
+func (plugin *azureDataDiskPlugin) GetVolumeLimits() (map[string]int64, error) {
+	cloud := plugin.host.GetCloudProvider()
+	if cloud.ProviderName() != azure.CloudProviderName {
+		return nil, fmt.Errorf("Expected Azure cloudprovider, got %s", cloud.ProviderName())
+	}
+
+	volumeLimits := map[string]int64{
+		util.AzureVolumeLimitKey: 16,
+	}
+	return volumeLimits, nil
+}
+
+func (plugin *azureDataDiskPlugin) VolumeLimitKey(spec *volume.Spec) string {
+	return util.AzureVolumeLimitKey
+}
+
 func (plugin *azureDataDiskPlugin) GetAccessModes() []v1.PersistentVolumeAccessMode {
 	return []v1.PersistentVolumeAccessMode{
 		v1.ReadWriteOnce,

pkg/volume/gce_pd/gce_pd.go

@@ -28,6 +28,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	gcecloud "k8s.io/kubernetes/pkg/cloudprovider/providers/gce"
 	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/util/mount"
 	kstrings "k8s.io/kubernetes/pkg/util/strings"
@@ -49,6 +50,7 @@ var _ volume.PersistentVolumePlugin = &gcePersistentDiskPlugin{}
 var _ volume.DeletableVolumePlugin = &gcePersistentDiskPlugin{}
 var _ volume.ProvisionableVolumePlugin = &gcePersistentDiskPlugin{}
 var _ volume.ExpandableVolumePlugin = &gcePersistentDiskPlugin{}
+var _ volume.VolumePluginWithAttachLimits = &gcePersistentDiskPlugin{}
 
 const (
 	gcePersistentDiskPluginName = "kubernetes.io/gce-pd"
@@ -100,6 +102,23 @@ func (plugin *gcePersistentDiskPlugin) GetAccessModes() []v1.PersistentVolumeAcc
 	}
 }
 
+func (plugin *gcePersistentDiskPlugin) GetVolumeLimits() (map[string]int64, error) {
+	cloud := plugin.host.GetCloudProvider()
+
+	if cloud.ProviderName() != gcecloud.ProviderName {
+		return nil, fmt.Errorf("Expected gce cloud got %s", cloud.ProviderName())
+	}
+
+	volumeLimits := map[string]int64{
+		util.GCEVolumeLimitKey: 16,
+	}
+	return volumeLimits, nil
+}
+
+func (plugin *gcePersistentDiskPlugin) VolumeLimitKey(spec *volume.Spec) string {
+	return util.GCEVolumeLimitKey
+}
+
 func (plugin *gcePersistentDiskPlugin) NewMounter(spec *volume.Spec, pod *v1.Pod, _ volume.VolumeOptions) (volume.Mounter, error) {
 	// Inject real implementations here, test through the internal function.
 	return plugin.newMounterInternal(spec, pod.UID, &GCEDiskUtil{}, plugin.host.GetMounter(plugin.GetPluginName()))

pkg/volume/plugins.go

@@ -216,6 +216,32 @@ type ExpandableVolumePlugin interface {
 	RequiresFSResize() bool
 }
 
+// VolumePluginWithAttachLimits is an extended interface of VolumePlugin that restricts number of
+// volumes that can be attached to a node.
+type VolumePluginWithAttachLimits interface {
+	VolumePlugin
+	// Return maximum number of volumes that can be attached to a node for this plugin.
+	// The key must be same as string returned by VolumeLimitKey function. The returned
+	// map may look like:
+	//     - { "storage-limits-aws-ebs": 39 }
+	//     - { "storage-limits-gce-pd": 10 }
+	// A volume plugin may return error from this function - if it can not be used on a given node or not
+	// applicable in given environment (where environment could be cloudprovider or any other dependency)
+	// For example - calling this function for EBS volume plugin on a GCE node should
+	// result in error.
+	// The returned values are stored in node allocatable property and will be used
+	// by scheduler to determine how many pods with volumes can be scheduled on given node.
+	GetVolumeLimits() (map[string]int64, error)
+	// Return volume limit key string to be used in node capacity constraints
+	// The key must start with prefix storage-limits-. For example:
+	//    - storage-limits-aws-ebs
+	//    - storage-limits-csi-cinder
+	// The key should respect character limit of ResourceName type
+	// This function may be called by kubelet or scheduler to identify node allocatable property
+	// which stores volumes limits.
+	VolumeLimitKey(spec *Spec) string
+}
+
 // BlockVolumePlugin is an extend interface of VolumePlugin and is used for block volumes support.
 type BlockVolumePlugin interface {
 	VolumePlugin
@@ -584,6 +610,17 @@ func (pm *VolumePluginMgr) refreshProbedPlugins() {
 	}
 }
 
+// ListVolumePluginWithLimits returns plugins that have volume limits on nodes
+func (pm *VolumePluginMgr) ListVolumePluginWithLimits() []VolumePluginWithAttachLimits {
+	matchedPlugins := []VolumePluginWithAttachLimits{}
+	for _, v := range pm.plugins {
+		if plugin, ok := v.(VolumePluginWithAttachLimits); ok {
+			matchedPlugins = append(matchedPlugins, plugin)
+		}
+	}
+	return matchedPlugins
+}
+
 // FindPersistentPluginBySpec looks for a persistent volume plugin that can
 // support a given volume specification.  If no plugin is found, return an
 // error
@@ -598,6 +635,20 @@ func (pm *VolumePluginMgr) FindPersistentPluginBySpec(spec *Spec) (PersistentVol
 	return nil, fmt.Errorf("no persistent volume plugin matched")
 }
 
+// FindVolumePluginWithLimitsBySpec returns volume plugin that has a limit on how many
+// of them can be attached to a node
+func (pm *VolumePluginMgr) FindVolumePluginWithLimitsBySpec(spec *Spec) (VolumePluginWithAttachLimits, error) {
+	volumePlugin, err := pm.FindPluginBySpec(spec)
+	if err != nil {
+		return nil, fmt.Errorf("Could not find volume plugin for spec : %#v", spec)
+	}
+
+	if limitedPlugin, ok := volumePlugin.(VolumePluginWithAttachLimits); ok {
+		return limitedPlugin, nil
+	}
+	return nil, fmt.Errorf("no plugin with limits found")
+}
+
 // FindPersistentPluginByName fetches a persistent volume plugin by name.  If
 // no plugin is found, returns error.
 func (pm *VolumePluginMgr) FindPersistentPluginByName(name string) (PersistentVolumePlugin, error) {

pkg/volume/util/BUILD

@@ -4,6 +4,7 @@ go_library(
     name = "go_default_library",
     srcs = [
         "atomic_writer.go",
+        "attach_limit.go",
         "device_util.go",
         "doc.go",
         "error.go",

pkg/volume/util/attach_limit.go

@@ -0,0 +1,29 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package util
+
+// This file is a common place holder for volume limit utility constants
+// shared between volume package and scheduler
+
+const (
+	// EBSVolumeLimitKey resource name that will store volume limits for EBS
+	EBSVolumeLimitKey = "attachable-volumes-aws-ebs"
+	// AzureVolumeLimitKey stores resource name that will store volume limits for Azure
+	AzureVolumeLimitKey = "attachable-volumes-azure-disk"
+	// GCEVolumeLimitKey stores resource name that will store volume limits for GCE node
+	GCEVolumeLimitKey = "attachable-volumes-gce-pd"
+)