github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-08-03 01:06:27 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #29006 from jsafrane/dynprov2

Browse Source 
Automatic merge from submit-queue

Implement dynamic provisioning (beta) of PersistentVolumes via StorageClass

Implemented according to PR #26908. There are several patches in this PR with one huge code regen inside.

* Please review the API changes (the first patch) carefully, sometimes I don't know what the code is doing...

* `PV.Spec.Class` and `PVC.Spec.Class` is not implemented, use annotation `volume.alpha.kubernetes.io/storage-class`

* See e2e test and integration test changes - Kubernetes won't provision a thing without explicit configuration of at least one `StorageClass` instance!

* Multiple provisioning volume plugins can coexist together, e.g. HostPath and AWS EBS. This is important for Gluster and RBD provisioners in #25026

* Contradicting the proposal, `claim.Selector` and `volume.alpha.kubernetes.io/storage-class` annotation are **not** mutually exclusive. They're both used for matching existing PVs. However, only `volume.alpha.kubernetes.io/storage-class` is used for provisioning, configuration of provisioning with `Selector` is left for (near) future.

* Documentation is missing. Can please someone write some while I am out?

For now, AWS volume plugin accepts classes with these parameters:

```
kind: StorageClass
metadata:
  name: slow
provisionerType: kubernetes.io/aws-ebs
provisionerParameters:
  type: io1
  zone: us-east-1d
  iopsPerGB: 10
```

* parameters are case-insensitive
* `type`: `io1`, `gp2`, `sc1`, `st1`. See AWS docs for details
* `iopsPerGB`: only for `io1` volumes. I/O operations per second per GiB. AWS volume plugin multiplies this with size of requested volume to compute IOPS of the volume and caps it at 20 000 IOPS (maximum supported by AWS, see AWS docs).
* of course, the plugin will use some defaults when a parameter is omitted in a `StorageClass` instance (`gp2` in the same zone as in 1.3).

GCE:

```
apiVersion: extensions/v1beta1
kind: StorageClass
metadata:
  name: slow
provisionerType: kubernetes.io/gce-pd
provisionerParameters:
  type: pd-standard
  zone: us-central1-a
```

* `type`: `pd-standard` or `pd-ssd`
* `zone`: GCE zone
* of course, the plugin will use some defaults when a parameter is omitted in a `StorageClass` instance (SSD in the same zone as in 1.3 ?).


No OpenStack/Cinder yet

@kubernetes/sig-storage
This commit is contained in:
Kubernetes Submit Queue 2016-08-18 09:56:16 -07:00 committed by GitHub
commit 9d2a5fe5e8
31 changed files with 1180 additions and 349 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
cmd/kube-controller-manager/app/controllermanager.go
View File

@ -419,19 +419,21 @@ func StartControllers(s *options.CMServer, kubeClient *client.Client, kubeconfig
glog.Infof("Not starting %s apis", groupVersion) glog.Infof("Not starting %s apis", groupVersion)
} }
provisioner, err := NewVolumeProvisioner(cloud, s.VolumeConfiguration) alphaProvisioner, err := NewAlphaVolumeProvisioner(cloud, s.VolumeConfiguration)
if err != nil { if err != nil {
glog.Fatalf("A Provisioner could not be created: %v, but one was expected. Provisioning will not work. This functionality is considered an early Alpha version.", err) glog.Fatalf("An backward-compatible provisioner could not be created: %v, but one was expected. Provisioning will not work. This functionality is considered an early Alpha version.", err)
} }
volumeController := persistentvolumecontroller.NewPersistentVolumeController( volumeController := persistentvolumecontroller.NewPersistentVolumeController(
clientset.NewForConfigOrDie(restclient.AddUserAgent(kubeconfig, "persistent-volume-binder")), clientset.NewForConfigOrDie(restclient.AddUserAgent(kubeconfig, "persistent-volume-binder")),
s.PVClaimBinderSyncPeriod.Duration, s.PVClaimBinderSyncPeriod.Duration,
provisioner, alphaProvisioner,
ProbeRecyclableVolumePlugins(s.VolumeConfiguration), ProbeControllerVolumePlugins(cloud, s.VolumeConfiguration),
cloud, cloud,
s.ClusterName, s.ClusterName,
nil, nil, nil, nil, // volumeSource
nil, // claimSource
nil, // classSource
nil, // eventRecorder
s.VolumeConfiguration.EnableDynamicProvisioning, s.VolumeConfiguration.EnableDynamicProvisioning,
) )
volumeController.Run() volumeController.Run()

39
cmd/kube-controller-manager/app/plugins.go
View File

@ -62,8 +62,10 @@ func ProbeAttachableVolumePlugins(config componentconfig.VolumeConfiguration) []
return allPlugins return allPlugins
} }
// ProbeRecyclableVolumePlugins collects all persistent volume plugins into an easy to use list. // ProbeControllerVolumePlugins collects all persistent volume plugins into an
func ProbeRecyclableVolumePlugins(config componentconfig.VolumeConfiguration) []volume.VolumePlugin { // easy to use list. Only volume plugins that implement any of
// provisioner/recycler/deleter interface should be returned.
func ProbeControllerVolumePlugins(cloud cloudprovider.Interface, config componentconfig.VolumeConfiguration) []volume.VolumePlugin {
allPlugins := []volume.VolumePlugin{} allPlugins := []volume.VolumePlugin{}
// The list of plugins to probe is decided by this binary, not // The list of plugins to probe is decided by this binary, not
@ -79,6 +81,7 @@ func ProbeRecyclableVolumePlugins(config componentconfig.VolumeConfiguration) []
RecyclerMinimumTimeout: int(config.PersistentVolumeRecyclerConfiguration.MinimumTimeoutHostPath), RecyclerMinimumTimeout: int(config.PersistentVolumeRecyclerConfiguration.MinimumTimeoutHostPath),
RecyclerTimeoutIncrement: int(config.PersistentVolumeRecyclerConfiguration.IncrementTimeoutHostPath), RecyclerTimeoutIncrement: int(config.PersistentVolumeRecyclerConfiguration.IncrementTimeoutHostPath),
RecyclerPodTemplate: volume.NewPersistentVolumeRecyclerPodTemplate(), RecyclerPodTemplate: volume.NewPersistentVolumeRecyclerPodTemplate(),
ProvisioningEnabled: config.EnableHostPathProvisioning,
} }
if err := AttemptToLoadRecycler(config.PersistentVolumeRecyclerConfiguration.PodTemplateFilePathHostPath, &hostPathConfig); err != nil { if err := AttemptToLoadRecycler(config.PersistentVolumeRecyclerConfiguration.PodTemplateFilePathHostPath, &hostPathConfig); err != nil {
glog.Fatalf("Could not create hostpath recycler pod from file %s: %+v", config.PersistentVolumeRecyclerConfiguration.PodTemplateFilePathHostPath, err) glog.Fatalf("Could not create hostpath recycler pod from file %s: %+v", config.PersistentVolumeRecyclerConfiguration.PodTemplateFilePathHostPath, err)
@ -95,22 +98,34 @@ func ProbeRecyclableVolumePlugins(config componentconfig.VolumeConfiguration) []
} }
allPlugins = append(allPlugins, nfs.ProbeVolumePlugins(nfsConfig)...) allPlugins = append(allPlugins, nfs.ProbeVolumePlugins(nfsConfig)...)
allPlugins = append(allPlugins, aws_ebs.ProbeVolumePlugins()...) if cloud != nil {
allPlugins = append(allPlugins, gce_pd.ProbeVolumePlugins()...) switch {
allPlugins = append(allPlugins, cinder.ProbeVolumePlugins()...) case aws.ProviderName == cloud.ProviderName():
allPlugins = append(allPlugins, vsphere_volume.ProbeVolumePlugins()...) allPlugins = append(allPlugins, aws_ebs.ProbeVolumePlugins()...)
case gce.ProviderName == cloud.ProviderName():
allPlugins = append(allPlugins, gce_pd.ProbeVolumePlugins()...)
case openstack.ProviderName == cloud.ProviderName():
allPlugins = append(allPlugins, cinder.ProbeVolumePlugins()...)
case vsphere.ProviderName == cloud.ProviderName():
allPlugins = append(allPlugins, vsphere_volume.ProbeVolumePlugins()...)
}
}
return allPlugins return allPlugins
} }
// NewVolumeProvisioner returns a volume provisioner to use when running in a cloud or development environment. // NewAlphaVolumeProvisioner returns a volume provisioner to use when running in
// The beta implementation of provisioning allows 1 implied provisioner per cloud, until we allow configuration of many. // a cloud or development environment. The alpha implementation of provisioning
// We explicitly map clouds to volume plugins here which allows us to configure many later without backwards compatibility issues. // allows 1 implied provisioner per cloud and is here only for compatibility
// Not all cloudproviders have provisioning capability, which is the reason for the bool in the return to tell the caller to expect one or not. // with Kubernetes 1.3
func NewVolumeProvisioner(cloud cloudprovider.Interface, config componentconfig.VolumeConfiguration) (volume.ProvisionableVolumePlugin, error) { // TODO: remove in Kubernetes 1.5
func NewAlphaVolumeProvisioner(cloud cloudprovider.Interface, config componentconfig.VolumeConfiguration) (volume.ProvisionableVolumePlugin, error) {
switch { switch {
case cloud == nil && config.EnableHostPathProvisioning: case cloud == nil && config.EnableHostPathProvisioning:
return getProvisionablePluginFromVolumePlugins(host_path.ProbeVolumePlugins(volume.VolumeConfig{})) return getProvisionablePluginFromVolumePlugins(host_path.ProbeVolumePlugins(
volume.VolumeConfig{
ProvisioningEnabled: true,
}))
case cloud != nil && aws.ProviderName == cloud.ProviderName(): case cloud != nil && aws.ProviderName == cloud.ProviderName():
return getProvisionablePluginFromVolumePlugins(aws_ebs.ProbeVolumePlugins()) return getProvisionablePluginFromVolumePlugins(aws_ebs.ProbeVolumePlugins())
case cloud != nil && gce.ProviderName == cloud.ProviderName(): case cloud != nil && gce.ProviderName == cloud.ProviderName():

16
contrib/mesos/pkg/controllermanager/controllermanager.go
View File

@ -285,21 +285,21 @@ func (s *CMServer) Run(_ []string) error {
} }
} }
provisioner, err := kubecontrollermanager.NewVolumeProvisioner(cloud, s.VolumeConfiguration) alphaProvisioner, err := kubecontrollermanager.NewAlphaVolumeProvisioner(cloud, s.VolumeConfiguration)
if err != nil { if err != nil {
glog.Fatalf("A Provisioner could not be created: %v, but one was expected. Provisioning will not work. This functionality is considered an early Alpha version.", err) glog.Fatalf("An backward-compatible provisioner could not be created: %v, but one was expected. Provisioning will not work. This functionality is considered an early Alpha version.", err)
} }
volumeController := persistentvolumecontroller.NewPersistentVolumeController( volumeController := persistentvolumecontroller.NewPersistentVolumeController(
clientset.NewForConfigOrDie(restclient.AddUserAgent(kubeconfig, "persistent-volume-binder")), clientset.NewForConfigOrDie(restclient.AddUserAgent(kubeconfig, "persistent-volume-binder")),
s.PVClaimBinderSyncPeriod.Duration, s.PVClaimBinderSyncPeriod.Duration,
provisioner, alphaProvisioner,
kubecontrollermanager.ProbeRecyclableVolumePlugins(s.VolumeConfiguration), kubecontrollermanager.ProbeControllerVolumePlugins(cloud, s.VolumeConfiguration),
cloud, cloud,
s.ClusterName, s.ClusterName,
nil, nil, // volumeSource
nil, nil, // claimSource
nil, nil, // classSource
nil, // eventRecorder
s.VolumeConfiguration.EnableDynamicProvisioning, s.VolumeConfiguration.EnableDynamicProvisioning,
) )
volumeController.Run() volumeController.Run()

20
docs/proposals/volume-provisioning.md
View File

@ -122,13 +122,13 @@ We propose that:
a match is found. The claim is `Pending` during this period. a match is found. The claim is `Pending` during this period.
4. With StorageClass instance, the controller finds volume plugin specified by 4. With StorageClass instance, the controller finds volume plugin specified by
StorageClass.ProvisionerType. StorageClass.Provisioner.
5. All provisioners are in-tree; they implement an interface called 5. All provisioners are in-tree; they implement an interface called
`ProvisionableVolumePlugin`, which has a method called `NewProvisioner` `ProvisionableVolumePlugin`, which has a method called `NewProvisioner`
that returns a new provisioner. that returns a new provisioner.
6. The controller calls volume plugin `Provision` with ProvisionerParameters from the `StorageClass` configuration object. 6. The controller calls volume plugin `Provision` with Parameters from the `StorageClass` configuration object.
7. If `Provision` returns an error, the controller generates an event on the 7. If `Provision` returns an error, the controller generates an event on the
claim and goes back to step 1., i.e. it will retry provisioning periodically claim and goes back to step 1., i.e. it will retry provisioning periodically
@ -166,11 +166,11 @@ type StorageClass struct {
unversioned.TypeMeta `json:",inline"` unversioned.TypeMeta `json:",inline"`
ObjectMeta `json:"metadata,omitempty"` ObjectMeta `json:"metadata,omitempty"`
// ProvisionerType indicates the type of the provisioner. // Provisioner indicates the type of the provisioner.
ProvisionerType string `json:"provisionerType,omitempty"` Provisioner string `json:"provisioner,omitempty"`
// Parameters for dynamic volume provisioner. // Parameters for dynamic volume provisioner.
ProvisionerParameters map[string]string `json:"provisionerParameters,omitempty"` Parameters map[string]string `json:"parameters,omitempty"`
} }
``` ```
@ -207,7 +207,7 @@ With the scheme outlined above the provisioner creates PVs using parameters spec
### Provisioner interface changes ### Provisioner interface changes
`struct volume.VolumeOptions` (containing parameters for a provisioner plugin) `struct volume.VolumeOptions` (containing parameters for a provisioner plugin)
will be extended to contain StorageClass.ProvisionerParameters. will be extended to contain StorageClass.Parameters.
The existing provisioner implementations will be modified to accept the StorageClass configuration object. The existing provisioner implementations will be modified to accept the StorageClass configuration object.
@ -229,8 +229,8 @@ apiVersion: v1
kind: StorageClass kind: StorageClass
metadata: metadata:
name: aws-fast name: aws-fast
provisionerType: kubernetes.io/aws-ebs provisioner: kubernetes.io/aws-ebs
provisionerParameters: parameters:
zone: us-east-1b zone: us-east-1b
type: ssd type: ssd
@ -239,8 +239,8 @@ apiVersion: v1
kind: StorageClass kind: StorageClass
metadata: metadata:
name: aws-slow name: aws-slow
provisionerType: kubernetes.io/aws-ebs provisioner: kubernetes.io/aws-ebs
provisionerParameters: parameters:
zone: us-east-1b zone: us-east-1b
type: spinning type: spinning
``` ```

73
examples/experimental/persistent-volume-provisioning/README.md
View File

@ -43,34 +43,49 @@ scripts that launch kube-controller-manager.
### Admin Configuration ### Admin Configuration
No configuration is required by the admin! 3 cloud providers will be provided in the alpha version The admin must define `StorageClass` objects that describe named "classes" of storage offered in a cluster. Different classes might map to arbitrary levels or policies determined by the admin. When configuring a `StorageClass` object for persistent volume provisioning, the admin will need to describe the type of provisioner to use and the parameters that will be used by the provisioner when it provisions a `PersistentVolume` belonging to the class.
of this feature: EBS, GCE, and Cinder.
When Kubernetes is running in one of those clouds, there will be an implied provisioner. The name of a StorageClass object is significant, and is how users can request a particular class, by specifying the name in their `PersistentVolumeClaim`. The `provisioner` field must be specified as it determines what volume plugin is used for provisioning PVs. 2 cloud providers will be provided in the beta version of this feature: EBS and GCE. The `parameters` field contains the parameters that describe volumes belonging to the storage class. Different parameters may be accepted depending on the `provisioner`. For example, the value `io1`, for the parameter `type`, and the parameter `iopsPerGB` are specific to EBS . When a parameter is omitted, some default is used.
There is no provisioner when running outside of any of those 3 cloud providers.
A fourth provisioner is included for testing and development only. It creates HostPath volumes, #### AWS
which will never work outside of a single node cluster. It is not supported in any way except for
local for testing and development. This provisioner may be used by passing
`--enable-hostpath-provisioner=true` to the controller manager while bringing it up.
When using the `hack/local_up_cluster.sh` script, this flag is turned off by default.
It may be turned on by setting the environment variable `ENABLE_HOSTPATH_PROVISIONER`
to true prior to running the script.
```yaml
kind: StorageClass
apiVersion: extensions/v1beta1
metadata:
name: slow
provisioner: kubernetes.io/aws-ebs
parameters:
type: io1
zone: us-east-1d
iopsPerGB: "10"
``` ```
env ENABLE_HOSTPATH_PROVISIONER=true hack/local-up-cluster.sh
* `type`: `io1`, `gp2`, `sc1`, `st1`. See AWS docs for details. Default: `gp2`.
* `zone`: AWS zone. If not specified, a random zone in the same region as controller-manager will be chosen.
* `iopsPerGB`: only for `io1` volumes. I/O operations per second per GiB. AWS volume plugin multiplies this with size of requested volume to compute IOPS of the volume and caps it at 20 000 IOPS (maximum supported by AWS, see AWS docs).
#### GCE
```yaml
kind: StorageClass
apiVersion: extensions/v1beta1
metadata:
name: slow
provisioner: kubernetes.io/gce-pd
parameters:
type: pd-standard
zone: us-central1-a
``` ```
* `type`: `pd-standard` or `pd-ssd`. Default: `pd-ssd`
* `zone`: GCE zone. If not specified, a random zone in the same region as controller-manager will be chosen.
### User provisioning requests ### User provisioning requests
Users request dynamically provisioned storage by including a storage class in their `PersistentVolumeClaim`. Users request dynamically provisioned storage by including a storage class in their `PersistentVolumeClaim`.
The annotation `volume.alpha.kubernetes.io/storage-class` is used to access this experimental feature. The annotation `volume.beta.kubernetes.io/storage-class` is used to access this experimental feature. It is required that this value matches the name of a `StorageClass` configured by the administrator.
In the future, admins will be able to define many storage classes. In the future, the storage class may remain in an annotation or become a field on the claim itself.
The storage class may remain in an annotation or become a field on the claim itself.
> The value of the storage-class annotation does not matter in the alpha version of this feature. There is
a single implied provisioner per cloud (which creates 1 kind of volume in the provider). The full version of the feature
will require that this value matches what is configured by the administrator.
``` ```
{ {
@ -79,7 +94,7 @@ will require that this value matches what is configured by the administrator.
"metadata": { "metadata": {
"name": "claim1", "name": "claim1",
"annotations": { "annotations": {
"volume.alpha.kubernetes.io/storage-class": "foo" "volume.beta.kubernetes.io/storage-class": "slow"
} }
}, },
"spec": { "spec": {
@ -97,26 +112,28 @@ will require that this value matches what is configured by the administrator.
### Sample output ### Sample output
This example uses HostPath but any provisioner would follow the same flow. This example uses GCE but any provisioner would follow the same flow.
First we note there are no Persistent Volumes in the cluster. After creating a claim, we see a new PV is created First we note there are no Persistent Volumes in the cluster. After creating a storage class and a claim including that storage class, we see a new PV is created
and automatically bound to the claim requesting storage. and automatically bound to the claim requesting storage.
``` ```
$ kubectl get pv $ kubectl get pv
$ kubectl create -f examples/experimental/persistent-volume-provisioning/gce-pd.yaml
storageclass "slow" created
$ kubectl create -f examples/experimental/persistent-volume-provisioning/claim1.json $ kubectl create -f examples/experimental/persistent-volume-provisioning/claim1.json
I1012 13:07:57.666759 22875 decoder.go:141] decoding stream as JSON
persistentvolumeclaim "claim1" created persistentvolumeclaim "claim1" created
$ kubectl get pv $ kubectl get pv
NAME LABELS CAPACITY ACCESSMODES STATUS CLAIM REASON AGE NAME CAPACITY ACCESSMODES STATUS CLAIM REASON AGE
pv-hostpath-r6z5o createdby=hostpath-dynamic-provisioner 3Gi RWO Bound default/claim1 2s pvc-bb6d2f0c-534c-11e6-9348-42010af00002 3Gi RWO Bound default/claim1 4s
$ kubectl get pvc $ kubectl get pvc
NAME LABELS STATUS VOLUME CAPACITY ACCESSMODES AGE NAME LABELS STATUS VOLUME CAPACITY ACCESSMODES AGE
claim1 <none> Bound pv-hostpath-r6z5o 3Gi RWO 7s claim1 <none> Bound pvc-bb6d2f0c-534c-11e6-9348-42010af00002 3Gi RWO 7s
# delete the claim to release the volume # delete the claim to release the volume
$ kubectl delete pvc claim1 $ kubectl delete pvc claim1

9
examples/experimental/persistent-volume-provisioning/aws-ebs.yaml Normal file
View File

@ -0,0 +1,9 @@
kind: StorageClass
apiVersion: extensions/v1beta1
metadata:
name: slow
provisioner: kubernetes.io/aws-ebs
parameters:
type: io1
zone: us-east-1d
iopsPerGB: "10"

2
examples/experimental/persistent-volume-provisioning/claim1.json
View File

@ -4,7 +4,7 @@
"metadata": { "metadata": {
"name": "claim1", "name": "claim1",
"annotations": { "annotations": {
"volume.alpha.kubernetes.io/storage-class": "foo" "volume.beta.kubernetes.io/storage-class": "slow"
} }
}, },
"spec": { "spec": {

20
examples/experimental/persistent-volume-provisioning/claim2.json
View File

@ -1,20 +0,0 @@
{
"kind": "PersistentVolumeClaim",
"apiVersion": "v1",
"metadata": {
"name": "claim2",
"annotations": {
"volume.alpha.kubernetes.io/storage-class": "bar"
}
},
"spec": {
"accessModes": [
"ReadWriteOnce"
],
"resources": {
"requests": {
"storage": "3Gi"
}
}
}
}

8
examples/experimental/persistent-volume-provisioning/gce-pd.yaml Normal file
View File

@ -0,0 +1,8 @@
kind: StorageClass
apiVersion: extensions/v1beta1
metadata:
name: slow
provisioner: kubernetes.io/gce-pd
parameters:
type: pd-standard
zone: us-central1-a

61
pkg/cloudprovider/providers/aws/aws.go
View File

@ -209,11 +209,29 @@ type EC2Metadata interface {
GetMetadata(path string) (string, error) GetMetadata(path string) (string, error)
} }
// AWS volume types
const (
// Provisioned IOPS SSD
VolumeTypeIO1 = "io1"
// General Purpose SSD
VolumeTypeGP2 = "gp2"
// Cold HDD (sc1)
VolumeTypeSC1 = "sc1"
// Throughput Optimized HDD
VolumeTypeST1 = "st1"
)
// VolumeOptions specifies capacity and tags for a volume. // VolumeOptions specifies capacity and tags for a volume.
type VolumeOptions struct { type VolumeOptions struct {
CapacityGB int CapacityGB int
Tags map[string]string Tags map[string]string
PVCName string PVCName string
VolumeType string
AvailabilityZone string
// IOPSPerGB x CapacityGB will give total IOPS of the volume to create.
// IOPSPerGB must be bigger than zero and smaller or equal to 30.
// Calculated total IOPS will be capped at 20000 IOPS.
IOPSPerGB int
} }
// Volumes is an interface for managing cloud-provisioned volumes // Volumes is an interface for managing cloud-provisioned volumes
@ -1475,14 +1493,47 @@ func (c *Cloud) CreateDisk(volumeOptions *VolumeOptions) (string, error) {
return "", fmt.Errorf("error querying for all zones: %v", err) return "", fmt.Errorf("error querying for all zones: %v", err)
} }
createAZ := volume.ChooseZoneForVolume(allZones, volumeOptions.PVCName) createAZ := volumeOptions.AvailabilityZone
if createAZ == "" {
createAZ = volume.ChooseZoneForVolume(allZones, volumeOptions.PVCName)
}
var createType string
var iops int64
switch volumeOptions.VolumeType {
case VolumeTypeGP2, VolumeTypeSC1, VolumeTypeST1:
createType = volumeOptions.VolumeType
case VolumeTypeIO1:
// See http://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_CreateVolume.html for IOPS constraints
if volumeOptions.IOPSPerGB <= 0 || volumeOptions.IOPSPerGB > 30 {
return "", fmt.Errorf("invalid iopsPerGB value %d, must be 0 < IOPSPerGB <= 30", volumeOptions.IOPSPerGB)
}
createType = volumeOptions.VolumeType
iops = int64(volumeOptions.CapacityGB * volumeOptions.IOPSPerGB)
if iops < 100 {
iops = 100
}
if iops > 20000 {
iops = 20000
}
case "":
createType = DefaultVolumeType
default:
return "", fmt.Errorf("invalid AWS VolumeType %q", volumeOptions.VolumeType)
}
// TODO: Should we tag this with the cluster id (so it gets deleted when the cluster does?) // TODO: Should we tag this with the cluster id (so it gets deleted when the cluster does?)
request := &ec2.CreateVolumeInput{} request := &ec2.CreateVolumeInput{}
request.AvailabilityZone = &createAZ request.AvailabilityZone = &createAZ
volSize := int64(volumeOptions.CapacityGB) volSize := int64(volumeOptions.CapacityGB)
request.Size = &volSize request.Size = &volSize
request.VolumeType = aws.String(DefaultVolumeType) request.VolumeType = &createType
if iops > 0 {
request.Iops = &iops
}
response, err := c.ec2.CreateVolume(request) response, err := c.ec2.CreateVolume(request)
if err != nil { if err != nil {
return "", err return "", err

27
pkg/cloudprovider/providers/gce/gce.go
View File

@ -101,6 +101,16 @@ type Config struct {
} }
} }
type DiskType string
const (
DiskTypeSSD = "pd-ssd"
DiskTypeStandard = "pd-standard"
diskTypeDefault = DiskTypeStandard
diskTypeUriTemplate = "https://www.googleapis.com/compute/v1/projects/%s/zones/%s/diskTypes/%s"
)
// Disks is interface for manipulation with GCE PDs. // Disks is interface for manipulation with GCE PDs.
type Disks interface { type Disks interface {
// AttachDisk attaches given disk to given instance. Current instance // AttachDisk attaches given disk to given instance. Current instance
@ -116,7 +126,7 @@ type Disks interface {
// CreateDisk creates a new PD with given properties. Tags are serialized // CreateDisk creates a new PD with given properties. Tags are serialized
// as JSON into Description field. // as JSON into Description field.
CreateDisk(name string, zone string, sizeGb int64, tags map[string]string) error CreateDisk(name string, diskType string, zone string, sizeGb int64, tags map[string]string) error
// DeleteDisk deletes PD. // DeleteDisk deletes PD.
DeleteDisk(diskToDelete string) error DeleteDisk(diskToDelete string) error
@ -2258,18 +2268,29 @@ func (gce *GCECloud) encodeDiskTags(tags map[string]string) (string, error) {
} }
// CreateDisk creates a new Persistent Disk, with the specified name & size, in // CreateDisk creates a new Persistent Disk, with the specified name & size, in
// the specified zone. It stores specified tags endoced in JSON in Description // the specified zone. It stores specified tags encoded in JSON in Description
// field. // field.
func (gce *GCECloud) CreateDisk(name string, zone string, sizeGb int64, tags map[string]string) error { func (gce *GCECloud) CreateDisk(name string, diskType string, zone string, sizeGb int64, tags map[string]string) error {
tagsStr, err := gce.encodeDiskTags(tags) tagsStr, err := gce.encodeDiskTags(tags)
if err != nil { if err != nil {
return err return err
} }
switch diskType {
case DiskTypeSSD, DiskTypeStandard:
// noop
case "":
diskType = diskTypeDefault
default:
return fmt.Errorf("invalid GCE disk type %q", diskType)
}
diskTypeUri := fmt.Sprintf(diskTypeUriTemplate, gce.projectID, zone, diskType)
diskToCreate := &compute.Disk{ diskToCreate := &compute.Disk{
Name: name, Name: name,
SizeGb: sizeGb, SizeGb: sizeGb,
Description: tagsStr, Description: tagsStr,
Type: diskTypeUri,
} }
createOp, err := gce.service.Disks.Insert(gce.projectID, zone, diskToCreate).Do() createOp, err := gce.service.Disks.Insert(gce.projectID, zone, diskToCreate).Do()

85
pkg/controller/volume/persistentvolume/binder_test.go
View File

@ -20,6 +20,7 @@ import (
"testing" "testing"
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/apis/extensions"
) )
// Test single call to syncClaim and syncVolume methods. // Test single call to syncClaim and syncVolume methods.
@ -421,8 +422,88 @@ func TestSync(t *testing.T) {
newClaimArray("claim4-8", "uid4-8", "10Gi", "volume4-8-x", api.ClaimBound), newClaimArray("claim4-8", "uid4-8", "10Gi", "volume4-8-x", api.ClaimBound),
noevents, noerrors, testSyncVolume, noevents, noerrors, testSyncVolume,
}, },
// PVC with class
{
// syncVolume binds a claim to requested class even if there is a
// smaller PV available
"13-1 - binding to class",
[]*api.PersistentVolume{
newVolume("volume13-1-1", "1Gi", "", "", api.VolumePending, api.PersistentVolumeReclaimRetain),
newVolume("volume13-1-2", "10Gi", "", "", api.VolumePending, api.PersistentVolumeReclaimRetain, annClass),
},
[]*api.PersistentVolume{
newVolume("volume13-1-1", "1Gi", "", "", api.VolumePending, api.PersistentVolumeReclaimRetain),
newVolume("volume13-1-2", "10Gi", "uid13-1", "claim13-1", api.VolumeBound, api.PersistentVolumeReclaimRetain, annBoundByController, annClass),
},
newClaimArray("claim13-1", "uid13-1", "1Gi", "", api.ClaimPending, annClass),
withExpectedCapacity("10Gi", newClaimArray("claim13-1", "uid13-1", "1Gi", "volume13-1-2", api.ClaimBound, annBoundByController, annClass, annBindCompleted)),
noevents, noerrors, testSyncClaim,
},
{
// syncVolume binds a claim without a class even if there is a
// smaller PV with a class available
"13-2 - binding without a class",
[]*api.PersistentVolume{
newVolume("volume13-2-1", "1Gi", "", "", api.VolumePending, api.PersistentVolumeReclaimRetain, annClass),
newVolume("volume13-2-2", "10Gi", "", "", api.VolumePending, api.PersistentVolumeReclaimRetain),
},
[]*api.PersistentVolume{
newVolume("volume13-2-1", "1Gi", "", "", api.VolumePending, api.PersistentVolumeReclaimRetain, annClass),
newVolume("volume13-2-2", "10Gi", "uid13-2", "claim13-2", api.VolumeBound, api.PersistentVolumeReclaimRetain, annBoundByController),
},
newClaimArray("claim13-2", "uid13-2", "1Gi", "", api.ClaimPending),
withExpectedCapacity("10Gi", newClaimArray("claim13-2", "uid13-2", "1Gi", "volume13-2-2", api.ClaimBound, annBoundByController, annBindCompleted)),
noevents, noerrors, testSyncClaim,
},
{
// syncVolume binds a claim with given class even if there is a
// smaller PV with different class available
"13-3 - binding to specific a class",
volumeWithClass("silver", []*api.PersistentVolume{
newVolume("volume13-3-1", "1Gi", "", "", api.VolumePending, api.PersistentVolumeReclaimRetain),
newVolume("volume13-3-2", "10Gi", "", "", api.VolumePending, api.PersistentVolumeReclaimRetain, annClass),
}),
volumeWithClass("silver", []*api.PersistentVolume{
newVolume("volume13-3-1", "1Gi", "", "", api.VolumePending, api.PersistentVolumeReclaimRetain),
newVolume("volume13-3-2", "10Gi", "uid13-3", "claim13-3", api.VolumeBound, api.PersistentVolumeReclaimRetain, annBoundByController, annClass),
}),
newClaimArray("claim13-3", "uid13-3", "1Gi", "", api.ClaimPending, annClass),
withExpectedCapacity("10Gi", newClaimArray("claim13-3", "uid13-3", "1Gi", "volume13-3-2", api.ClaimBound, annBoundByController, annBindCompleted, annClass)),
noevents, noerrors, testSyncClaim,
},
{
// syncVolume binds claim requesting class "" to claim to PV with
// class=""
"13-4 - empty class",
volumeWithClass("", newVolumeArray("volume13-4", "1Gi", "", "", api.VolumePending, api.PersistentVolumeReclaimRetain)),
volumeWithClass("", newVolumeArray("volume13-4", "1Gi", "uid13-4", "claim13-4", api.VolumeBound, api.PersistentVolumeReclaimRetain, annBoundByController)),
claimWithClass("", newClaimArray("claim13-4", "uid13-4", "1Gi", "", api.ClaimPending)),
claimWithClass("", newClaimArray("claim13-4", "uid13-4", "1Gi", "volume13-4", api.ClaimBound, annBoundByController, annBindCompleted)),
noevents, noerrors, testSyncClaim,
},
{
// syncVolume binds claim requesting class nil to claim to PV with
// class = ""
"13-5 - nil class",
volumeWithClass("", newVolumeArray("volume13-5", "1Gi", "", "", api.VolumePending, api.PersistentVolumeReclaimRetain)),
volumeWithClass("", newVolumeArray("volume13-5", "1Gi", "uid13-5", "claim13-5", api.VolumeBound, api.PersistentVolumeReclaimRetain, annBoundByController)),
newClaimArray("claim13-5", "uid13-5", "1Gi", "", api.ClaimPending),
newClaimArray("claim13-5", "uid13-5", "1Gi", "volume13-5", api.ClaimBound, annBoundByController, annBindCompleted),
noevents, noerrors, testSyncClaim,
},
{
// syncVolume binds claim requesting class "" to claim to PV with
// class=nil
"13-6 - nil class in PV, '' class in claim",
newVolumeArray("volume13-6", "1Gi", "", "", api.VolumePending, api.PersistentVolumeReclaimRetain),
newVolumeArray("volume13-6", "1Gi", "uid13-6", "claim13-6", api.VolumeBound, api.PersistentVolumeReclaimRetain, annBoundByController),
claimWithClass("", newClaimArray("claim13-6", "uid13-6", "1Gi", "", api.ClaimPending)),
claimWithClass("", newClaimArray("claim13-6", "uid13-6", "1Gi", "volume13-6", api.ClaimBound, annBoundByController, annBindCompleted)),
noevents, noerrors, testSyncClaim,
},
} }
runSyncTests(t, tests) runSyncTests(t, tests, []*extensions.StorageClass{})
} }
// Test multiple calls to syncClaim/syncVolume and periodic sync of all // Test multiple calls to syncClaim/syncVolume and periodic sync of all
@ -469,5 +550,5 @@ func TestMultiSync(t *testing.T) {
}, },
} }
runMultisyncTests(t, tests) runMultisyncTests(t, tests, []*extensions.StorageClass{}, "")
} }

158
pkg/controller/volume/persistentvolume/controller.go
View File

@ -22,6 +22,8 @@ import (
"time" "time"
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/unversioned"
"k8s.io/kubernetes/pkg/apis/extensions"
"k8s.io/kubernetes/pkg/client/cache" "k8s.io/kubernetes/pkg/client/cache"
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset" clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
"k8s.io/kubernetes/pkg/client/record" "k8s.io/kubernetes/pkg/client/record"
@ -108,10 +110,19 @@ const annBindCompleted = "pv.kubernetes.io/bind-completed"
// pre-bound). Value of this annotation does not matter. // pre-bound). Value of this annotation does not matter.
const annBoundByController = "pv.kubernetes.io/bound-by-controller" const annBoundByController = "pv.kubernetes.io/bound-by-controller"
// annClass annotation represents a new field which instructs dynamic // annClass annotation represents the storage class associated with a resource:
// provisioning to choose a particular storage class (aka profile). // - in PersistentVolumeClaim it represents required class to match.
// Value of this annotation should be empty. // Only PersistentVolumes with the same class (i.e. annotation with the same
const annClass = "volume.alpha.kubernetes.io/storage-class" // value) can be bound to the claim. In case no such volume exists, the
// controller will provision a new one using StorageClass instance with
// the same name as the annotation value.
// - in PersistentVolume it represents storage class to which the persistent
// volume belongs.
const annClass = "volume.beta.kubernetes.io/storage-class"
// alphaAnnClass annotation represents the previous alpha storage class
// annotation. it's no longer used and held here for posterity.
const annAlphaClass = "volume.alpha.kubernetes.io/storage-class"
// This annotation is added to a PV that has been dynamically provisioned by // This annotation is added to a PV that has been dynamically provisioned by
// Kubernetes. Its value is name of volume plugin that created the volume. // Kubernetes. Its value is name of volume plugin that created the volume.
@ -148,11 +159,13 @@ type PersistentVolumeController struct {
claimController *framework.Controller claimController *framework.Controller
claimControllerStopCh chan struct{} claimControllerStopCh chan struct{}
claimSource cache.ListerWatcher claimSource cache.ListerWatcher
classReflector *cache.Reflector
classReflectorStopCh chan struct{}
classSource cache.ListerWatcher
kubeClient clientset.Interface kubeClient clientset.Interface
eventRecorder record.EventRecorder eventRecorder record.EventRecorder
cloud cloudprovider.Interface cloud cloudprovider.Interface
recyclePluginMgr vol.VolumePluginMgr volumePluginMgr vol.VolumePluginMgr
provisioner vol.ProvisionableVolumePlugin
enableDynamicProvisioning bool enableDynamicProvisioning bool
clusterName string clusterName string
@ -163,6 +176,7 @@ type PersistentVolumeController struct {
// it saves newer version to etcd. // it saves newer version to etcd.
volumes persistentVolumeOrderedIndex volumes persistentVolumeOrderedIndex
claims cache.Store claims cache.Store
classes cache.Store
// Map of scheduled/running operations. // Map of scheduled/running operations.
runningOperations goroutinemap.GoRoutineMap runningOperations goroutinemap.GoRoutineMap
@ -173,6 +187,10 @@ type PersistentVolumeController struct {
createProvisionedPVRetryCount int createProvisionedPVRetryCount int
createProvisionedPVInterval time.Duration createProvisionedPVInterval time.Duration
// Provisioner for annAlphaClass.
// TODO: remove in 1.5
alphaProvisioner vol.ProvisionableVolumePlugin
} }
// syncClaim is the main controller method to decide what to do with a claim. // syncClaim is the main controller method to decide what to do with a claim.
@ -198,6 +216,7 @@ func (ctrl *PersistentVolumeController) syncUnboundClaim(claim *api.PersistentVo
// OBSERVATION: pvc is "Pending" // OBSERVATION: pvc is "Pending"
if claim.Spec.VolumeName == "" { if claim.Spec.VolumeName == "" {
// User did not care which PV they get. // User did not care which PV they get.
// [Unit test set 1] // [Unit test set 1]
volume, err := ctrl.volumes.findBestMatchForClaim(claim) volume, err := ctrl.volumes.findBestMatchForClaim(claim)
if err != nil { if err != nil {
@ -208,7 +227,8 @@ func (ctrl *PersistentVolumeController) syncUnboundClaim(claim *api.PersistentVo
glog.V(4).Infof("synchronizing unbound PersistentVolumeClaim[%s]: no volume found", claimToClaimKey(claim)) glog.V(4).Infof("synchronizing unbound PersistentVolumeClaim[%s]: no volume found", claimToClaimKey(claim))
// No PV could be found // No PV could be found
// OBSERVATION: pvc is "Pending", will retry // OBSERVATION: pvc is "Pending", will retry
if hasAnnotation(claim.ObjectMeta, annClass) { // TODO: remove Alpha check in 1.5
if getClaimClass(claim) != "" || hasAnnotation(claim.ObjectMeta, annAlphaClass) {
if err = ctrl.provisionClaim(claim); err != nil { if err = ctrl.provisionClaim(claim); err != nil {
return err return err
} }
@ -965,7 +985,7 @@ func (ctrl *PersistentVolumeController) recycleVolumeOperation(arg interface{})
// Find a plugin. // Find a plugin.
spec := vol.NewSpecFromPersistentVolume(volume, false) spec := vol.NewSpecFromPersistentVolume(volume, false)
plugin, err := ctrl.recyclePluginMgr.FindRecyclablePluginBySpec(spec) plugin, err := ctrl.volumePluginMgr.FindRecyclablePluginBySpec(spec)
if err != nil { if err != nil {
// No recycler found. Emit an event and mark the volume Failed. // No recycler found. Emit an event and mark the volume Failed.
if _, err = ctrl.updateVolumePhaseWithEvent(volume, api.VolumeFailed, api.EventTypeWarning, "VolumeFailedRecycle", "No recycler plugin found for the volume!"); err != nil { if _, err = ctrl.updateVolumePhaseWithEvent(volume, api.VolumeFailed, api.EventTypeWarning, "VolumeFailedRecycle", "No recycler plugin found for the volume!"); err != nil {
@ -1119,14 +1139,33 @@ func (ctrl *PersistentVolumeController) isVolumeReleased(volume *api.PersistentV
// (it will be re-used in future provisioner error cases). // (it will be re-used in future provisioner error cases).
func (ctrl *PersistentVolumeController) doDeleteVolume(volume *api.PersistentVolume) error { func (ctrl *PersistentVolumeController) doDeleteVolume(volume *api.PersistentVolume) error {
glog.V(4).Infof("doDeleteVolume [%s]", volume.Name) glog.V(4).Infof("doDeleteVolume [%s]", volume.Name)
// Find a plugin. var err error
spec := vol.NewSpecFromPersistentVolume(volume, false)
plugin, err := ctrl.recyclePluginMgr.FindDeletablePluginBySpec(spec) // Find a plugin. Try to find the same plugin that provisioned the volume
if err != nil { var plugin vol.DeletableVolumePlugin
// No deleter found. Emit an event and mark the volume Failed. if hasAnnotation(volume.ObjectMeta, annDynamicallyProvisioned) {
return fmt.Errorf("Error getting deleter volume plugin for volume %q: %v", volume.Name, err) provisionPluginName := volume.Annotations[annDynamicallyProvisioned]
if provisionPluginName != "" {
plugin, err = ctrl.volumePluginMgr.FindDeletablePluginByName(provisionPluginName)
if err != nil {
glog.V(3).Infof("did not find a deleter plugin %q for volume %q: %v, will try to find a generic one",
provisionPluginName, volume.Name, err)
}
}
} }
spec := vol.NewSpecFromPersistentVolume(volume, false)
if plugin == nil {
// The plugin that provisioned the volume was not found or the volume
// was not dynamically provisioned. Try to find a plugin by spec.
plugin, err = ctrl.volumePluginMgr.FindDeletablePluginBySpec(spec)
if err != nil {
// No deleter found. Emit an event and mark the volume Failed.
return fmt.Errorf("Error getting deleter volume plugin for volume %q: %v", volume.Name, err)
}
}
glog.V(5).Infof("found a deleter plugin %q for volume %q", plugin.GetPluginName(), volume.Name)
// Plugin found // Plugin found
deleter, err := plugin.NewDeleter(spec) deleter, err := plugin.NewDeleter(spec)
if err != nil { if err != nil {
@ -1143,7 +1182,8 @@ func (ctrl *PersistentVolumeController) doDeleteVolume(volume *api.PersistentVol
return nil return nil
} }
// provisionClaim starts new asynchronous operation to provision a claim if provisioning is enabled. // provisionClaim starts new asynchronous operation to provision a claim if
// provisioning is enabled.
func (ctrl *PersistentVolumeController) provisionClaim(claim *api.PersistentVolumeClaim) error { func (ctrl *PersistentVolumeController) provisionClaim(claim *api.PersistentVolumeClaim) error {
if !ctrl.enableDynamicProvisioning { if !ctrl.enableDynamicProvisioning {
return nil return nil
@ -1165,7 +1205,9 @@ func (ctrl *PersistentVolumeController) provisionClaimOperation(claimObj interfa
glog.Errorf("Cannot convert provisionClaimOperation argument to claim, got %#v", claimObj) glog.Errorf("Cannot convert provisionClaimOperation argument to claim, got %#v", claimObj)
return return
} }
glog.V(4).Infof("provisionClaimOperation [%s] started", claimToClaimKey(claim))
claimClass := getClaimClass(claim)
glog.V(4).Infof("provisionClaimOperation [%s] started, class: %q", claimToClaimKey(claim), claimClass)
// A previous doProvisionClaim may just have finished while we were waiting for // A previous doProvisionClaim may just have finished while we were waiting for
// the locks. Check that PV (with deterministic name) hasn't been provisioned // the locks. Check that PV (with deterministic name) hasn't been provisioned
@ -1187,12 +1229,10 @@ func (ctrl *PersistentVolumeController) provisionClaimOperation(claimObj interfa
return return
} }
// TODO: find provisionable plugin based on a class/profile plugin, storageClass, err := ctrl.findProvisionablePlugin(claim)
plugin := ctrl.provisioner if err != nil {
if plugin == nil { ctrl.eventRecorder.Event(claim, api.EventTypeWarning, "ProvisioningFailed", err.Error())
// No provisioner found. Emit an event. glog.V(2).Infof("error finding provisioning plugin for claim %s: %v", claimToClaimKey(claim), err)
ctrl.eventRecorder.Event(claim, api.EventTypeWarning, "ProvisioningFailed", "No provisioner plugin found for the claim!")
glog.V(2).Infof("no provisioner plugin found for claim %s!", claimToClaimKey(claim))
// The controller will retry provisioning the volume in every // The controller will retry provisioning the volume in every
// syncVolume() call. // syncVolume() call.
return return
@ -1212,21 +1252,23 @@ func (ctrl *PersistentVolumeController) provisionClaimOperation(claimObj interfa
ClusterName: ctrl.clusterName, ClusterName: ctrl.clusterName,
PVName: pvName, PVName: pvName,
PVCName: claim.Name, PVCName: claim.Name,
Parameters: storageClass.Parameters,
Selector: claim.Spec.Selector,
} }
// Provision the volume // Provision the volume
provisioner, err := plugin.NewProvisioner(options) provisioner, err := plugin.NewProvisioner(options)
if err != nil { if err != nil {
strerr := fmt.Sprintf("Failed to create provisioner: %v", err) strerr := fmt.Sprintf("Failed to create provisioner: %v", err)
glog.V(2).Infof("failed to create provisioner for claim %q: %v", claimToClaimKey(claim), err) glog.V(2).Infof("failed to create provisioner for claim %q with StorageClass %q: %v", claimToClaimKey(claim), storageClass.Name, err)
ctrl.eventRecorder.Event(claim, api.EventTypeWarning, "ProvisioningFailed", strerr) ctrl.eventRecorder.Event(claim, api.EventTypeWarning, "ProvisioningFailed", strerr)
return return
} }
volume, err = provisioner.Provision() volume, err = provisioner.Provision()
if err != nil { if err != nil {
strerr := fmt.Sprintf("Failed to provision volume: %v", err) strerr := fmt.Sprintf("Failed to provision volume with StorageClass %q: %v", storageClass.Name, err)
glog.V(2).Infof("failed to provision volume for claim %q: %v", claimToClaimKey(claim), err) glog.V(2).Infof("failed to provision volume for claim %q with StorageClass %q: %v", claimToClaimKey(claim), storageClass.Name, err)
ctrl.eventRecorder.Event(claim, api.EventTypeWarning, "ProvisioningFailed", strerr) ctrl.eventRecorder.Event(claim, api.EventTypeWarning, "ProvisioningFailed", strerr)
return return
} }
@ -1242,6 +1284,12 @@ func (ctrl *PersistentVolumeController) provisionClaimOperation(claimObj interfa
// Add annBoundByController (used in deleting the volume) // Add annBoundByController (used in deleting the volume)
setAnnotation(&volume.ObjectMeta, annBoundByController, "yes") setAnnotation(&volume.ObjectMeta, annBoundByController, "yes")
setAnnotation(&volume.ObjectMeta, annDynamicallyProvisioned, plugin.GetPluginName()) setAnnotation(&volume.ObjectMeta, annDynamicallyProvisioned, plugin.GetPluginName())
// For Alpha provisioning behavior, do not add annClass for volumes created
// by annAlphaClass
// TODO: remove this check in 1.5, annClass will be always non-empty there.
if claimClass != "" {
setAnnotation(&volume.ObjectMeta, annClass, claimClass)
}
// Try to create the PV object several times // Try to create the PV object several times
for i := 0; i < ctrl.createProvisionedPVRetryCount; i++ { for i := 0; i < ctrl.createProvisionedPVRetryCount; i++ {
@ -1320,3 +1368,63 @@ func (ctrl *PersistentVolumeController) scheduleOperation(operationName string,
} }
} }
} }
func (ctrl *PersistentVolumeController) findProvisionablePlugin(claim *api.PersistentVolumeClaim) (vol.ProvisionableVolumePlugin, *extensions.StorageClass, error) {
// TODO: remove this alpha behavior in 1.5
alpha := hasAnnotation(claim.ObjectMeta, annAlphaClass)
beta := hasAnnotation(claim.ObjectMeta, annClass)
if alpha && beta {
// Both Alpha and Beta annotations are set. Do beta.
alpha = false
msg := fmt.Sprintf("both %q and %q annotations are present, using %q", annAlphaClass, annClass, annClass)
ctrl.eventRecorder.Event(claim, api.EventTypeNormal, "ProvisioningIgnoreAlpha", msg)
}
if alpha {
// Fall back to fixed list of provisioner plugins
return ctrl.findAlphaProvisionablePlugin()
}
// provisionClaim() which leads here is never called with claimClass=="", we
// can save some checks.
claimClass := getClaimClass(claim)
classObj, found, err := ctrl.classes.GetByKey(claimClass)
if err != nil {
return nil, nil, err
}
if !found {
return nil, nil, fmt.Errorf("StorageClass %q not found", claimClass)
}
class, ok := classObj.(*extensions.StorageClass)
if !ok {
return nil, nil, fmt.Errorf("Cannot convert object to StorageClass: %+v", classObj)
}
// Find a plugin for the class
plugin, err := ctrl.volumePluginMgr.FindProvisionablePluginByName(class.Provisioner)
if err != nil {
return nil, nil, err
}
return plugin, class, nil
}
// findAlphaProvisionablePlugin returns a volume plugin compatible with
// Kubernetes 1.3.
// TODO: remove in Kubernetes 1.5
func (ctrl *PersistentVolumeController) findAlphaProvisionablePlugin() (vol.ProvisionableVolumePlugin, *extensions.StorageClass, error) {
if ctrl.alphaProvisioner == nil {
return nil, nil, fmt.Errorf("cannot find volume plugin for alpha provisioning")
}
// Return a dummy StorageClass instance with no parameters
storageClass := &extensions.StorageClass{
TypeMeta: unversioned.TypeMeta{
Kind: "StorageClass",
},
ObjectMeta: api.ObjectMeta{
Name: "",
},
Provisioner: ctrl.alphaProvisioner.GetPluginName(),
}
glog.V(4).Infof("using alpha provisioner %s", ctrl.alphaProvisioner.GetPluginName())
return ctrl.alphaProvisioner, storageClass, nil
}

83
pkg/controller/volume/persistentvolume/controller_base.go
View File

@ -24,6 +24,7 @@ import (
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/errors" "k8s.io/kubernetes/pkg/api/errors"
"k8s.io/kubernetes/pkg/api/meta" "k8s.io/kubernetes/pkg/api/meta"
"k8s.io/kubernetes/pkg/apis/extensions"
"k8s.io/kubernetes/pkg/client/cache" "k8s.io/kubernetes/pkg/client/cache"
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset" clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
unversioned_core "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/core/unversioned" unversioned_core "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/core/unversioned"
@ -47,11 +48,11 @@ import (
func NewPersistentVolumeController( func NewPersistentVolumeController(
kubeClient clientset.Interface, kubeClient clientset.Interface,
syncPeriod time.Duration, syncPeriod time.Duration,
provisioner vol.ProvisionableVolumePlugin, alphaProvisioner vol.ProvisionableVolumePlugin,
recyclers []vol.VolumePlugin, volumePlugins []vol.VolumePlugin,
cloud cloudprovider.Interface, cloud cloudprovider.Interface,
clusterName string, clusterName string,
volumeSource, claimSource cache.ListerWatcher, volumeSource, claimSource, classSource cache.ListerWatcher,
eventRecorder record.EventRecorder, eventRecorder record.EventRecorder,
enableDynamicProvisioning bool, enableDynamicProvisioning bool,
) *PersistentVolumeController { ) *PersistentVolumeController {
@ -63,23 +64,23 @@ func NewPersistentVolumeController(
} }
controller := &PersistentVolumeController{ controller := &PersistentVolumeController{
volumes: newPersistentVolumeOrderedIndex(), volumes: newPersistentVolumeOrderedIndex(),
claims: cache.NewStore(framework.DeletionHandlingMetaNamespaceKeyFunc), claims: cache.NewStore(framework.DeletionHandlingMetaNamespaceKeyFunc),
kubeClient: kubeClient, kubeClient: kubeClient,
eventRecorder: eventRecorder, eventRecorder: eventRecorder,
runningOperations: goroutinemap.NewGoRoutineMap(false /* exponentialBackOffOnError */), runningOperations: goroutinemap.NewGoRoutineMap(false /* exponentialBackOffOnError */),
cloud: cloud, cloud: cloud,
provisioner: provisioner,
enableDynamicProvisioning: enableDynamicProvisioning, enableDynamicProvisioning: enableDynamicProvisioning,
clusterName: clusterName, clusterName: clusterName,
createProvisionedPVRetryCount: createProvisionedPVRetryCount, createProvisionedPVRetryCount: createProvisionedPVRetryCount,
createProvisionedPVInterval: createProvisionedPVInterval, createProvisionedPVInterval: createProvisionedPVInterval,
alphaProvisioner: alphaProvisioner,
} }
controller.recyclePluginMgr.InitPlugins(recyclers, controller) controller.volumePluginMgr.InitPlugins(volumePlugins, controller)
if controller.provisioner != nil { if controller.alphaProvisioner != nil {
if err := controller.provisioner.Init(controller); err != nil { if err := controller.alphaProvisioner.Init(controller); err != nil {
glog.Errorf("PersistentVolumeController: error initializing provisioner plugin: %v", err) glog.Errorf("PersistentVolumeController: error initializing alpha provisioner plugin: %v", err)
} }
} }
@ -107,6 +108,18 @@ func NewPersistentVolumeController(
} }
controller.claimSource = claimSource controller.claimSource = claimSource
if classSource == nil {
classSource = &cache.ListWatch{
ListFunc: func(options api.ListOptions) (runtime.Object, error) {
return kubeClient.Extensions().StorageClasses().List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
return kubeClient.Extensions().StorageClasses().Watch(options)
},
}
}
controller.classSource = classSource
_, controller.volumeController = framework.NewIndexerInformer( _, controller.volumeController = framework.NewIndexerInformer(
volumeSource, volumeSource,
&api.PersistentVolume{}, &api.PersistentVolume{},
@ -128,6 +141,16 @@ func NewPersistentVolumeController(
DeleteFunc: controller.deleteClaim, DeleteFunc: controller.deleteClaim,
}, },
) )
// This is just a cache of StorageClass instances, no special actions are
// needed when a class is created/deleted/updated.
controller.classes = cache.NewStore(framework.DeletionHandlingMetaNamespaceKeyFunc)
controller.classReflector = cache.NewReflector(
classSource,
&extensions.StorageClass{},
controller.classes,
syncPeriod,
)
return controller return controller
} }
@ -433,6 +456,11 @@ func (ctrl *PersistentVolumeController) Run() {
ctrl.claimControllerStopCh = make(chan struct{}) ctrl.claimControllerStopCh = make(chan struct{})
go ctrl.claimController.Run(ctrl.claimControllerStopCh) go ctrl.claimController.Run(ctrl.claimControllerStopCh)
} }
if ctrl.classReflectorStopCh == nil {
ctrl.classReflectorStopCh = make(chan struct{})
go ctrl.classReflector.RunUntil(ctrl.classReflectorStopCh)
}
} }
// Stop gracefully shuts down this controller // Stop gracefully shuts down this controller
@ -440,6 +468,7 @@ func (ctrl *PersistentVolumeController) Stop() {
glog.V(4).Infof("stopping PersistentVolumeController") glog.V(4).Infof("stopping PersistentVolumeController")
close(ctrl.volumeControllerStopCh) close(ctrl.volumeControllerStopCh)
close(ctrl.claimControllerStopCh) close(ctrl.claimControllerStopCh)
close(ctrl.classReflectorStopCh)
} }
const ( const (
@ -578,3 +607,29 @@ func storeObjectUpdate(store cache.Store, obj interface{}, className string) (bo
} }
return true, nil return true, nil
} }
// getVolumeClass returns value of annClass annotation or empty string in case
// the annotation does not exist.
// TODO: change to PersistentVolume.Spec.Class value when this attribute is
// introduced.
func getVolumeClass(volume *api.PersistentVolume) string {
if class, found := volume.Annotations[annClass]; found {
return class
}
// 'nil' is interpreted as "", i.e. the volume does not belong to any class.
return ""
}
// getClaimClass returns name of class that is requested by given claim.
// Request for `nil` class is interpreted as request for class "",
// i.e. for a classless PV.
func getClaimClass(claim *api.PersistentVolumeClaim) string {
// TODO: change to PersistentVolumeClaim.Spec.Class value when this
// attribute is introduced.
if class, found := claim.Annotations[annClass]; found {
return class
}
return ""
}

2
pkg/controller/volume/persistentvolume/controller_test.go
View File

@ -163,7 +163,7 @@ func TestControllerSync(t *testing.T) {
client := &fake.Clientset{} client := &fake.Clientset{}
volumeSource := framework.NewFakePVControllerSource() volumeSource := framework.NewFakePVControllerSource()
claimSource := framework.NewFakePVCControllerSource() claimSource := framework.NewFakePVCControllerSource()
ctrl := newTestController(client, volumeSource, claimSource, true) ctrl := newTestController(client, volumeSource, claimSource, nil, true)
reactor := newVolumeReactor(client, ctrl, volumeSource, claimSource, test.errors) reactor := newVolumeReactor(client, ctrl, volumeSource, claimSource, test.errors)
for _, claim := range test.initialClaims { for _, claim := range test.initialClaims {
claimSource.Add(claim) claimSource.Add(claim)

21
pkg/controller/volume/persistentvolume/delete_test.go
View File

@ -21,6 +21,7 @@ import (
"testing" "testing"
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/apis/extensions"
) )
// Test single call to syncVolume, expecting recycling to happen. // Test single call to syncVolume, expecting recycling to happen.
@ -39,7 +40,7 @@ func TestDeleteSync(t *testing.T) {
noevents, noerrors, noevents, noerrors,
// Inject deleter into the controller and call syncVolume. The // Inject deleter into the controller and call syncVolume. The
// deleter simulates one delete() call that succeeds. // deleter simulates one delete() call that succeeds.
wrapTestWithControllerConfig(operationDelete, []error{nil}, testSyncVolume), wrapTestWithReclaimCalls(operationDelete, []error{nil}, testSyncVolume),
}, },
{ {
// delete volume bound by user // delete volume bound by user
@ -51,7 +52,7 @@ func TestDeleteSync(t *testing.T) {
noevents, noerrors, noevents, noerrors,
// Inject deleter into the controller and call syncVolume. The // Inject deleter into the controller and call syncVolume. The
// deleter simulates one delete() call that succeeds. // deleter simulates one delete() call that succeeds.
wrapTestWithControllerConfig(operationDelete, []error{nil}, testSyncVolume), wrapTestWithReclaimCalls(operationDelete, []error{nil}, testSyncVolume),
}, },
{ {
// delete failure - plugin not found // delete failure - plugin not found
@ -70,7 +71,7 @@ func TestDeleteSync(t *testing.T) {
noclaims, noclaims,
noclaims, noclaims,
[]string{"Warning VolumeFailedDelete"}, noerrors, []string{"Warning VolumeFailedDelete"}, noerrors,
wrapTestWithControllerConfig(operationDelete, []error{}, testSyncVolume), wrapTestWithReclaimCalls(operationDelete, []error{}, testSyncVolume),
}, },
{ {
// delete failure - delete() returns error // delete failure - delete() returns error
@ -80,7 +81,7 @@ func TestDeleteSync(t *testing.T) {
noclaims, noclaims,
noclaims, noclaims,
[]string{"Warning VolumeFailedDelete"}, noerrors, []string{"Warning VolumeFailedDelete"}, noerrors,
wrapTestWithControllerConfig(operationDelete, []error{errors.New("Mock delete error")}, testSyncVolume), wrapTestWithReclaimCalls(operationDelete, []error{errors.New("Mock delete error")}, testSyncVolume),
}, },
{ {
// delete success(?) - volume is deleted before doDelete() starts // delete success(?) - volume is deleted before doDelete() starts
@ -90,7 +91,7 @@ func TestDeleteSync(t *testing.T) {
noclaims, noclaims,
noclaims, noclaims,
noevents, noerrors, noevents, noerrors,
wrapTestWithInjectedOperation(wrapTestWithControllerConfig(operationDelete, []error{}, testSyncVolume), func(ctrl *PersistentVolumeController, reactor *volumeReactor) { wrapTestWithInjectedOperation(wrapTestWithReclaimCalls(operationDelete, []error{}, testSyncVolume), func(ctrl *PersistentVolumeController, reactor *volumeReactor) {
// Delete the volume before delete operation starts // Delete the volume before delete operation starts
reactor.lock.Lock() reactor.lock.Lock()
delete(reactor.volumes, "volume8-6") delete(reactor.volumes, "volume8-6")
@ -107,7 +108,7 @@ func TestDeleteSync(t *testing.T) {
noclaims, noclaims,
newClaimArray("claim8-7", "uid8-7", "10Gi", "volume8-7", api.ClaimBound), newClaimArray("claim8-7", "uid8-7", "10Gi", "volume8-7", api.ClaimBound),
noevents, noerrors, noevents, noerrors,
wrapTestWithInjectedOperation(wrapTestWithControllerConfig(operationDelete, []error{}, testSyncVolume), func(ctrl *PersistentVolumeController, reactor *volumeReactor) { wrapTestWithInjectedOperation(wrapTestWithReclaimCalls(operationDelete, []error{}, testSyncVolume), func(ctrl *PersistentVolumeController, reactor *volumeReactor) {
reactor.lock.Lock() reactor.lock.Lock()
defer reactor.lock.Unlock() defer reactor.lock.Unlock()
// Bind the volume to resurrected claim (this should never // Bind the volume to resurrected claim (this should never
@ -130,10 +131,10 @@ func TestDeleteSync(t *testing.T) {
noevents, noerrors, noevents, noerrors,
// Inject deleter into the controller and call syncVolume. The // Inject deleter into the controller and call syncVolume. The
// deleter simulates one delete() call that succeeds. // deleter simulates one delete() call that succeeds.
wrapTestWithControllerConfig(operationDelete, []error{nil}, testSyncVolume), wrapTestWithReclaimCalls(operationDelete, []error{nil}, testSyncVolume),
}, },
} }
runSyncTests(t, tests) runSyncTests(t, tests, []*extensions.StorageClass{})
} }
// Test multiple calls to syncClaim/syncVolume and periodic sync of all // Test multiple calls to syncClaim/syncVolume and periodic sync of all
@ -161,9 +162,9 @@ func TestDeleteMultiSync(t *testing.T) {
noclaims, noclaims,
noclaims, noclaims,
[]string{"Warning VolumeFailedDelete"}, noerrors, []string{"Warning VolumeFailedDelete"}, noerrors,
wrapTestWithControllerConfig(operationDelete, []error{errors.New("Mock delete error"), nil}, testSyncVolume), wrapTestWithReclaimCalls(operationDelete, []error{errors.New("Mock delete error"), nil}, testSyncVolume),
}, },
} }
runMultisyncTests(t, tests) runMultisyncTests(t, tests, []*extensions.StorageClass{}, "")
} }

161
pkg/controller/volume/persistentvolume/framework_test.go
View File

@ -33,6 +33,7 @@ import (
"k8s.io/kubernetes/pkg/api/resource" "k8s.io/kubernetes/pkg/api/resource"
"k8s.io/kubernetes/pkg/api/testapi" "k8s.io/kubernetes/pkg/api/testapi"
"k8s.io/kubernetes/pkg/api/unversioned" "k8s.io/kubernetes/pkg/api/unversioned"
"k8s.io/kubernetes/pkg/apis/extensions"
"k8s.io/kubernetes/pkg/client/cache" "k8s.io/kubernetes/pkg/client/cache"
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset" clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/fake" "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/fake"
@ -583,22 +584,26 @@ func newVolumeReactor(client *fake.Clientset, ctrl *PersistentVolumeController,
return reactor return reactor
} }
func newTestController(kubeClient clientset.Interface, volumeSource, claimSource cache.ListerWatcher, enableDynamicProvisioning bool) *PersistentVolumeController { func newTestController(kubeClient clientset.Interface, volumeSource, claimSource, classSource cache.ListerWatcher, enableDynamicProvisioning bool) *PersistentVolumeController {
if volumeSource == nil { if volumeSource == nil {
volumeSource = framework.NewFakePVControllerSource() volumeSource = framework.NewFakePVControllerSource()
} }
if claimSource == nil { if claimSource == nil {
claimSource = framework.NewFakePVCControllerSource() claimSource = framework.NewFakePVCControllerSource()
} }
if classSource == nil {
classSource = framework.NewFakeControllerSource()
}
ctrl := NewPersistentVolumeController( ctrl := NewPersistentVolumeController(
kubeClient, kubeClient,
5*time.Second, // sync period 5*time.Second, // sync period
nil, // provisioner nil, // alpha provisioner
[]vol.VolumePlugin{}, // recyclers []vol.VolumePlugin{}, // recyclers
nil, // cloud nil, // cloud
"", "",
volumeSource, volumeSource,
claimSource, claimSource,
classSource,
record.NewFakeRecorder(1000), // event recorder record.NewFakeRecorder(1000), // event recorder
enableDynamicProvisioning, enableDynamicProvisioning,
) )
@ -608,27 +613,6 @@ func newTestController(kubeClient clientset.Interface, volumeSource, claimSource
return ctrl return ctrl
} }
func addRecyclePlugin(ctrl *PersistentVolumeController, expectedRecycleCalls []error) {
plugin := &mockVolumePlugin{
recycleCalls: expectedRecycleCalls,
}
ctrl.recyclePluginMgr.InitPlugins([]vol.VolumePlugin{plugin}, ctrl)
}
func addDeletePlugin(ctrl *PersistentVolumeController, expectedDeleteCalls []error) {
plugin := &mockVolumePlugin{
deleteCalls: expectedDeleteCalls,
}
ctrl.recyclePluginMgr.InitPlugins([]vol.VolumePlugin{plugin}, ctrl)
}
func addProvisionPlugin(ctrl *PersistentVolumeController, expectedDeleteCalls []error) {
plugin := &mockVolumePlugin{
provisionCalls: expectedDeleteCalls,
}
ctrl.provisioner = plugin
}
// newVolume returns a new volume with given attributes // newVolume returns a new volume with given attributes
func newVolume(name, capacity, boundToClaimUID, boundToClaimName string, phase api.PersistentVolumePhase, reclaimPolicy api.PersistentVolumeReclaimPolicy, annotations ...string) *api.PersistentVolume { func newVolume(name, capacity, boundToClaimUID, boundToClaimName string, phase api.PersistentVolumePhase, reclaimPolicy api.PersistentVolumeReclaimPolicy, annotations ...string) *api.PersistentVolume {
volume := api.PersistentVolume{ volume := api.PersistentVolume{
@ -664,10 +648,13 @@ func newVolume(name, capacity, boundToClaimUID, boundToClaimName string, phase a
if len(annotations) > 0 { if len(annotations) > 0 {
volume.Annotations = make(map[string]string) volume.Annotations = make(map[string]string)
for _, a := range annotations { for _, a := range annotations {
if a != annDynamicallyProvisioned { switch a {
volume.Annotations[a] = "yes" case annDynamicallyProvisioned:
} else {
volume.Annotations[a] = mockPluginName volume.Annotations[a] = mockPluginName
case annClass:
volume.Annotations[a] = "gold"
default:
volume.Annotations[a] = "yes"
} }
} }
} }
@ -713,6 +700,17 @@ func withMessage(message string, volumes []*api.PersistentVolume) []*api.Persist
return volumes return volumes
} }
// volumeWithClass saves given class into annClass annotation.
// Meant to be used to compose claims specified inline in a test.
func volumeWithClass(className string, volumes []*api.PersistentVolume) []*api.PersistentVolume {
if volumes[0].Annotations == nil {
volumes[0].Annotations = map[string]string{annClass: className}
} else {
volumes[0].Annotations[annClass] = className
}
return volumes
}
// newVolumeArray returns array with a single volume that would be returned by // newVolumeArray returns array with a single volume that would be returned by
// newVolume() with the same parameters. // newVolume() with the same parameters.
func newVolumeArray(name, capacity, boundToClaimUID, boundToClaimName string, phase api.PersistentVolumePhase, reclaimPolicy api.PersistentVolumeReclaimPolicy, annotations ...string) []*api.PersistentVolume { func newVolumeArray(name, capacity, boundToClaimUID, boundToClaimName string, phase api.PersistentVolumePhase, reclaimPolicy api.PersistentVolumeReclaimPolicy, annotations ...string) []*api.PersistentVolume {
@ -749,7 +747,12 @@ func newClaim(name, claimUID, capacity, boundToVolume string, phase api.Persiste
if len(annotations) > 0 { if len(annotations) > 0 {
claim.Annotations = make(map[string]string) claim.Annotations = make(map[string]string)
for _, a := range annotations { for _, a := range annotations {
claim.Annotations[a] = "yes" switch a {
case annClass:
claim.Annotations[a] = "gold"
default:
claim.Annotations[a] = "yes"
}
} }
} }
@ -772,6 +775,17 @@ func newClaimArray(name, claimUID, capacity, boundToVolume string, phase api.Per
} }
} }
// claimWithClass saves given class into annClass annotation.
// Meant to be used to compose claims specified inline in a test.
func claimWithClass(className string, claims []*api.PersistentVolumeClaim) []*api.PersistentVolumeClaim {
if claims[0].Annotations == nil {
claims[0].Annotations = map[string]string{annClass: className}
} else {
claims[0].Annotations[annClass] = className
}
return claims
}
func testSyncClaim(ctrl *PersistentVolumeController, reactor *volumeReactor, test controllerTest) error { func testSyncClaim(ctrl *PersistentVolumeController, reactor *volumeReactor, test controllerTest) error {
return ctrl.syncClaim(test.initialClaims[0]) return ctrl.syncClaim(test.initialClaims[0])
} }
@ -793,29 +807,47 @@ type operationType string
const operationDelete = "Delete" const operationDelete = "Delete"
const operationRecycle = "Recycle" const operationRecycle = "Recycle"
const operationProvision = "Provision"
// wrapTestWithControllerConfig returns a testCall that: // wrapTestWithPluginCalls returns a testCall that:
// - configures controller with recycler, deleter or provisioner which will // - configures controller with a volume plugin that implements recycler,
// return provided errors when a volume is deleted, recycled or provisioned // deleter and provisioner. The plugin retunrs provided errors when a volume
// is deleted, recycled or provisioned.
// - calls given testCall // - calls given testCall
func wrapTestWithControllerConfig(operation operationType, expectedOperationCalls []error, toWrap testCall) testCall { func wrapTestWithPluginCalls(expectedRecycleCalls, expectedDeleteCalls []error, expectedProvisionCalls []provisionCall, toWrap testCall) testCall {
expected := expectedOperationCalls
return func(ctrl *PersistentVolumeController, reactor *volumeReactor, test controllerTest) error { return func(ctrl *PersistentVolumeController, reactor *volumeReactor, test controllerTest) error {
switch operation { plugin := &mockVolumePlugin{
case operationDelete: recycleCalls: expectedRecycleCalls,
addDeletePlugin(ctrl, expected) deleteCalls: expectedDeleteCalls,
case operationRecycle: provisionCalls: expectedProvisionCalls,
addRecyclePlugin(ctrl, expected) }
case operationProvision: ctrl.volumePluginMgr.InitPlugins([]vol.VolumePlugin{plugin}, ctrl)
addProvisionPlugin(ctrl, expected) if expectedProvisionCalls != nil {
ctrl.alphaProvisioner = plugin
} }
return toWrap(ctrl, reactor, test) return toWrap(ctrl, reactor, test)
} }
} }
// wrapTestWithReclaimCalls returns a testCall that:
// - configures controller with recycler or deleter which will return provided
// errors when a volume is deleted or recycled
// - calls given testCall
func wrapTestWithReclaimCalls(operation operationType, expectedOperationCalls []error, toWrap testCall) testCall {
if operation == operationDelete {
return wrapTestWithPluginCalls(nil, expectedOperationCalls, nil, toWrap)
} else {
return wrapTestWithPluginCalls(expectedOperationCalls, nil, nil, toWrap)
}
}
// wrapTestWithProvisionCalls returns a testCall that:
// - configures controller with a provisioner which will return provided errors
// when a claim is provisioned
// - calls given testCall
func wrapTestWithProvisionCalls(expectedProvisionCalls []provisionCall, toWrap testCall) testCall {
return wrapTestWithPluginCalls(nil, nil, expectedProvisionCalls, toWrap)
}
// wrapTestWithInjectedOperation returns a testCall that: // wrapTestWithInjectedOperation returns a testCall that:
// - starts the controller and lets it run original testCall until // - starts the controller and lets it run original testCall until
// scheduleOperation() call. It blocks the controller there and calls the // scheduleOperation() call. It blocks the controller there and calls the
@ -873,13 +905,13 @@ func evaluateTestResults(ctrl *PersistentVolumeController, reactor *volumeReacto
// 2. Call the tested function (syncClaim/syncVolume) via // 2. Call the tested function (syncClaim/syncVolume) via
// controllerTest.testCall *once*. // controllerTest.testCall *once*.
// 3. Compare resulting volumes and claims with expected volumes and claims. // 3. Compare resulting volumes and claims with expected volumes and claims.
func runSyncTests(t *testing.T, tests []controllerTest) { func runSyncTests(t *testing.T, tests []controllerTest, storageClasses []*extensions.StorageClass) {
for _, test := range tests { for _, test := range tests {
glog.V(4).Infof("starting test %q", test.name) glog.V(4).Infof("starting test %q", test.name)
// Initialize the controller // Initialize the controller
client := &fake.Clientset{} client := &fake.Clientset{}
ctrl := newTestController(client, nil, nil, true) ctrl := newTestController(client, nil, nil, nil, true)
reactor := newVolumeReactor(client, ctrl, nil, nil, test.errors) reactor := newVolumeReactor(client, ctrl, nil, nil, test.errors)
for _, claim := range test.initialClaims { for _, claim := range test.initialClaims {
ctrl.claims.Add(claim) ctrl.claims.Add(claim)
@ -890,6 +922,15 @@ func runSyncTests(t *testing.T, tests []controllerTest) {
reactor.volumes[volume.Name] = volume reactor.volumes[volume.Name] = volume
} }
// Convert classes to []interface{} and forcefully inject them into
// controller.
storageClassPtrs := make([]interface{}, len(storageClasses))
for i, s := range storageClasses {
storageClassPtrs[i] = s
}
// 1 is the resource version
ctrl.classes.Replace(storageClassPtrs, "1")
// Run the tested functions // Run the tested functions
err := test.test(ctrl, reactor, test) err := test.test(ctrl, reactor, test)
if err != nil { if err != nil {
@ -920,13 +961,22 @@ func runSyncTests(t *testing.T, tests []controllerTest) {
// 5. When 3. does not do any changes, finish the tests and compare final set // 5. When 3. does not do any changes, finish the tests and compare final set
// of volumes/claims with expected claims/volumes and report differences. // of volumes/claims with expected claims/volumes and report differences.
// Some limit of calls in enforced to prevent endless loops. // Some limit of calls in enforced to prevent endless loops.
func runMultisyncTests(t *testing.T, tests []controllerTest) { func runMultisyncTests(t *testing.T, tests []controllerTest, storageClasses []*extensions.StorageClass, defaultStorageClass string) {
for _, test := range tests { for _, test := range tests {
glog.V(4).Infof("starting multisync test %q", test.name) glog.V(4).Infof("starting multisync test %q", test.name)
// Initialize the controller // Initialize the controller
client := &fake.Clientset{} client := &fake.Clientset{}
ctrl := newTestController(client, nil, nil, true) ctrl := newTestController(client, nil, nil, nil, true)
// Convert classes to []interface{} and forcefully inject them into
// controller.
storageClassPtrs := make([]interface{}, len(storageClasses))
for i, s := range storageClasses {
storageClassPtrs[i] = s
}
ctrl.classes.Replace(storageClassPtrs, "1")
reactor := newVolumeReactor(client, ctrl, nil, nil, test.errors) reactor := newVolumeReactor(client, ctrl, nil, nil, test.errors)
for _, claim := range test.initialClaims { for _, claim := range test.initialClaims {
ctrl.claims.Add(claim) ctrl.claims.Add(claim)
@ -1022,7 +1072,7 @@ func runMultisyncTests(t *testing.T, tests []controllerTest) {
// Dummy volume plugin for provisioning, deletion and recycling. It contains // Dummy volume plugin for provisioning, deletion and recycling. It contains
// lists of expected return values to simulate errors. // lists of expected return values to simulate errors.
type mockVolumePlugin struct { type mockVolumePlugin struct {
provisionCalls []error provisionCalls []provisionCall
provisionCallCounter int provisionCallCounter int
deleteCalls []error deleteCalls []error
deleteCallCounter int deleteCallCounter int
@ -1031,6 +1081,11 @@ type mockVolumePlugin struct {
provisionOptions vol.VolumeOptions provisionOptions vol.VolumeOptions
} }
type provisionCall struct {
expectedParameters map[string]string
ret error
}
var _ vol.VolumePlugin = &mockVolumePlugin{} var _ vol.VolumePlugin = &mockVolumePlugin{}
var _ vol.RecyclableVolumePlugin = &mockVolumePlugin{} var _ vol.RecyclableVolumePlugin = &mockVolumePlugin{}
var _ vol.DeletableVolumePlugin = &mockVolumePlugin{} var _ vol.DeletableVolumePlugin = &mockVolumePlugin{}
@ -1087,8 +1142,12 @@ func (plugin *mockVolumePlugin) Provision() (*api.PersistentVolume, error) {
} }
var pv *api.PersistentVolume var pv *api.PersistentVolume
err := plugin.provisionCalls[plugin.provisionCallCounter] call := plugin.provisionCalls[plugin.provisionCallCounter]
if err == nil { if !reflect.DeepEqual(call.expectedParameters, plugin.provisionOptions.Parameters) {
glog.Errorf("invalid provisioner call, expected options: %+v, got: %+v", call.expectedParameters, plugin.provisionOptions.Parameters)
return nil, fmt.Errorf("Mock plugin error: invalid provisioner call")
}
if call.ret == nil {
// Create a fake PV with known GCE volume (to match expected volume) // Create a fake PV with known GCE volume (to match expected volume)
pv = &api.PersistentVolume{ pv = &api.PersistentVolume{
ObjectMeta: api.ObjectMeta{ ObjectMeta: api.ObjectMeta{
@ -1108,8 +1167,8 @@ func (plugin *mockVolumePlugin) Provision() (*api.PersistentVolume, error) {
} }
plugin.provisionCallCounter++ plugin.provisionCallCounter++
glog.V(4).Infof("mock plugin Provision call nr. %d, returning %v: %v", plugin.provisionCallCounter, pv, err) glog.V(4).Infof("mock plugin Provision call nr. %d, returning %v: %v", plugin.provisionCallCounter, pv, call.ret)
return pv, err return pv, call.ret
} }
// Deleter interfaces // Deleter interfaces

24
pkg/controller/volume/persistentvolume/index.go
View File

@ -92,6 +92,7 @@ func (pvIndex *persistentVolumeOrderedIndex) findByClaim(claim *api.PersistentVo
var smallestVolumeSize int64 var smallestVolumeSize int64
requestedQty := claim.Spec.Resources.Requests[api.ResourceName(api.ResourceStorage)] requestedQty := claim.Spec.Resources.Requests[api.ResourceName(api.ResourceStorage)]
requestedSize := requestedQty.Value() requestedSize := requestedQty.Value()
requestedClass := getClaimClass(claim)
var selector labels.Selector var selector labels.Selector
if claim.Spec.Selector != nil { if claim.Spec.Selector != nil {
@ -123,14 +124,26 @@ func (pvIndex *persistentVolumeOrderedIndex) findByClaim(claim *api.PersistentVo
return volume, nil return volume, nil
} }
// In Alpha dynamic provisioning, we do now want not match claims
// with existing PVs, findByClaim must find only PVs that are
// pre-bound to the claim (by dynamic provisioning). TODO: remove in
// 1.5
if hasAnnotation(claim.ObjectMeta, annAlphaClass) {
continue
}
// filter out: // filter out:
// - volumes bound to another claim // - volumes bound to another claim
// - volumes whose labels don't match the claim's selector, if specified // - volumes whose labels don't match the claim's selector, if specified
// - volumes in Class that is not requested
if volume.Spec.ClaimRef != nil { if volume.Spec.ClaimRef != nil {
continue continue
} else if selector != nil && !selector.Matches(labels.Set(volume.Labels)) { } else if selector != nil && !selector.Matches(labels.Set(volume.Labels)) {
continue continue
} }
if getVolumeClass(volume) != requestedClass {
continue
}
volumeQty := volume.Spec.Capacity[api.ResourceStorage] volumeQty := volume.Spec.Capacity[api.ResourceStorage]
volumeSize := volumeQty.Value() volumeSize := volumeQty.Value()
@ -142,17 +155,6 @@ func (pvIndex *persistentVolumeOrderedIndex) findByClaim(claim *api.PersistentVo
} }
} }
// We want to provision volumes if the annotation is set even if there
// is matching PV. Therefore, do not look for available PV and let
// a new volume to be provisioned.
//
// When provisioner creates a new PV to this claim, an exact match
// pre-bound to the claim will be found by the checks above during
// subsequent claim sync.
if hasAnnotation(claim.ObjectMeta, annClass) {
return nil, nil
}
if smallestVolume != nil { if smallestVolume != nil {
// Found a matching volume // Found a matching volume
return smallestVolume, nil return smallestVolume, nil

110
pkg/controller/volume/persistentvolume/index_test.go
View File

@ -164,7 +164,52 @@ func TestMatchVolume(t *testing.T) {
AccessModes: []api.PersistentVolumeAccessMode{api.ReadWriteOnce}, AccessModes: []api.PersistentVolumeAccessMode{api.ReadWriteOnce},
Resources: api.ResourceRequirements{ Resources: api.ResourceRequirements{
Requests: api.ResourceList{ Requests: api.ResourceList{
api.ResourceName(api.ResourceStorage): resource.MustParse("10000G"), api.ResourceName(api.ResourceStorage): resource.MustParse("20000G"),
},
},
},
},
},
"successful-match-with-class": {
expectedMatch: "gce-pd-silver1",
claim: &api.PersistentVolumeClaim{
ObjectMeta: api.ObjectMeta{
Name: "claim01",
Namespace: "myns",
Annotations: map[string]string{
annClass: "silver",
},
},
Spec: api.PersistentVolumeClaimSpec{
AccessModes: []api.PersistentVolumeAccessMode{api.ReadWriteOnce},
Selector: &unversioned.LabelSelector{
MatchLabels: map[string]string{
"should-exist": "true",
},
},
Resources: api.ResourceRequirements{
Requests: api.ResourceList{
api.ResourceName(api.ResourceStorage): resource.MustParse("1G"),
},
},
},
},
},
"successful-match-with-class-and-labels": {
expectedMatch: "gce-pd-silver2",
claim: &api.PersistentVolumeClaim{
ObjectMeta: api.ObjectMeta{
Name: "claim01",
Namespace: "myns",
Annotations: map[string]string{
annClass: "silver",
},
},
Spec: api.PersistentVolumeClaimSpec{
AccessModes: []api.PersistentVolumeAccessMode{api.ReadWriteOnce},
Resources: api.ResourceRequirements{
Requests: api.ResourceList{
api.ResourceName(api.ResourceStorage): resource.MustParse("1G"),
}, },
}, },
}, },
@ -561,6 +606,29 @@ func createTestVolumes() []*api.PersistentVolume {
"should-exist": "true", "should-exist": "true",
}, },
}, },
Spec: api.PersistentVolumeSpec{
Capacity: api.ResourceList{
api.ResourceName(api.ResourceStorage): resource.MustParse("20000G"),
},
PersistentVolumeSource: api.PersistentVolumeSource{
GCEPersistentDisk: &api.GCEPersistentDiskVolumeSource{},
},
AccessModes: []api.PersistentVolumeAccessMode{
api.ReadWriteOnce,
},
},
},
{
ObjectMeta: api.ObjectMeta{
UID: "gce-pd-silver1",
Name: "gce0023",
Labels: map[string]string{
"should-exist": "true",
},
Annotations: map[string]string{
annClass: "silver",
},
},
Spec: api.PersistentVolumeSpec{ Spec: api.PersistentVolumeSpec{
Capacity: api.ResourceList{ Capacity: api.ResourceList{
api.ResourceName(api.ResourceStorage): resource.MustParse("10000G"), api.ResourceName(api.ResourceStorage): resource.MustParse("10000G"),
@ -573,6 +641,46 @@ func createTestVolumes() []*api.PersistentVolume {
}, },
}, },
}, },
{
ObjectMeta: api.ObjectMeta{
UID: "gce-pd-silver2",
Name: "gce0024",
Annotations: map[string]string{
annClass: "silver",
},
},
Spec: api.PersistentVolumeSpec{
Capacity: api.ResourceList{
api.ResourceName(api.ResourceStorage): resource.MustParse("100G"),
},
PersistentVolumeSource: api.PersistentVolumeSource{
GCEPersistentDisk: &api.GCEPersistentDiskVolumeSource{},
},
AccessModes: []api.PersistentVolumeAccessMode{
api.ReadWriteOnce,
},
},
},
{
ObjectMeta: api.ObjectMeta{
UID: "gce-pd-gold",
Name: "gce0025",
Annotations: map[string]string{
annClass: "gold",
},
},
Spec: api.PersistentVolumeSpec{
Capacity: api.ResourceList{
api.ResourceName(api.ResourceStorage): resource.MustParse("50G"),
},
PersistentVolumeSource: api.PersistentVolumeSource{
GCEPersistentDisk: &api.GCEPersistentDiskVolumeSource{},
},
AccessModes: []api.PersistentVolumeAccessMode{
api.ReadWriteOnce,
},
},
},
} }
} }

198
pkg/controller/volume/persistentvolume/provision_test.go
View File

@ -21,8 +21,63 @@ import (
"testing" "testing"
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/unversioned"
"k8s.io/kubernetes/pkg/apis/extensions"
) )
var class1Parameters = map[string]string{
"param1": "value1",
}
var class2Parameters = map[string]string{
"param2": "value2",
}
var storageClasses = []*extensions.StorageClass{
{
TypeMeta: unversioned.TypeMeta{
Kind: "StorageClass",
},
ObjectMeta: api.ObjectMeta{
Name: "gold",
},
Provisioner: mockPluginName,
Parameters: class1Parameters,
},
{
TypeMeta: unversioned.TypeMeta{
Kind: "StorageClass",
},
ObjectMeta: api.ObjectMeta{
Name: "silver",
},
Provisioner: mockPluginName,
Parameters: class2Parameters,
},
}
// call to storageClass 1, returning an error
var provision1Error = provisionCall{
ret: errors.New("Moc provisioner error"),
expectedParameters: class1Parameters,
}
// call to storageClass 1, returning a valid PV
var provision1Success = provisionCall{
ret: nil,
expectedParameters: class1Parameters,
}
// call to storageClass 2, returning a valid PV
var provision2Success = provisionCall{
ret: nil,
expectedParameters: class2Parameters,
}
var provisionAlphaSuccess = provisionCall{
ret: nil,
}
// Test single call to syncVolume, expecting provisioning to happen. // Test single call to syncVolume, expecting provisioning to happen.
// 1. Fill in the controller with initial data // 1. Fill in the controller with initial data
// 2. Call the syncVolume *once*. // 2. Call the syncVolume *once*.
@ -30,14 +85,14 @@ import (
func TestProvisionSync(t *testing.T) { func TestProvisionSync(t *testing.T) {
tests := []controllerTest{ tests := []controllerTest{
{ {
// Provision a volume // Provision a volume (with a default class)
"11-1 - successful provision", "11-1 - successful provision with storage class 1",
novolumes, novolumes,
newVolumeArray("pvc-uid11-1", "1Gi", "uid11-1", "claim11-1", api.VolumeBound, api.PersistentVolumeReclaimDelete, annBoundByController, annDynamicallyProvisioned), newVolumeArray("pvc-uid11-1", "1Gi", "uid11-1", "claim11-1", api.VolumeBound, api.PersistentVolumeReclaimDelete, annBoundByController, annDynamicallyProvisioned, annClass),
newClaimArray("claim11-1", "uid11-1", "1Gi", "", api.ClaimPending, annClass), newClaimArray("claim11-1", "uid11-1", "1Gi", "", api.ClaimPending, annClass),
// Binding will be completed in the next syncClaim // Binding will be completed in the next syncClaim
newClaimArray("claim11-1", "uid11-1", "1Gi", "", api.ClaimPending, annClass), newClaimArray("claim11-1", "uid11-1", "1Gi", "", api.ClaimPending, annClass),
noevents, noerrors, wrapTestWithControllerConfig(operationProvision, []error{nil}, testSyncClaim), noevents, noerrors, wrapTestWithProvisionCalls([]provisionCall{provision1Success}, testSyncClaim),
}, },
{ {
// Provision failure - plugin not found // Provision failure - plugin not found
@ -57,7 +112,7 @@ func TestProvisionSync(t *testing.T) {
newClaimArray("claim11-3", "uid11-3", "1Gi", "", api.ClaimPending, annClass), newClaimArray("claim11-3", "uid11-3", "1Gi", "", api.ClaimPending, annClass),
newClaimArray("claim11-3", "uid11-3", "1Gi", "", api.ClaimPending, annClass), newClaimArray("claim11-3", "uid11-3", "1Gi", "", api.ClaimPending, annClass),
[]string{"Warning ProvisioningFailed"}, noerrors, []string{"Warning ProvisioningFailed"}, noerrors,
wrapTestWithControllerConfig(operationProvision, []error{}, testSyncClaim), wrapTestWithProvisionCalls([]provisionCall{}, testSyncClaim),
}, },
{ {
// Provision failure - Provision returns error // Provision failure - Provision returns error
@ -67,40 +122,35 @@ func TestProvisionSync(t *testing.T) {
newClaimArray("claim11-4", "uid11-4", "1Gi", "", api.ClaimPending, annClass), newClaimArray("claim11-4", "uid11-4", "1Gi", "", api.ClaimPending, annClass),
newClaimArray("claim11-4", "uid11-4", "1Gi", "", api.ClaimPending, annClass), newClaimArray("claim11-4", "uid11-4", "1Gi", "", api.ClaimPending, annClass),
[]string{"Warning ProvisioningFailed"}, noerrors, []string{"Warning ProvisioningFailed"}, noerrors,
wrapTestWithControllerConfig(operationProvision, []error{errors.New("Moc provisioner error")}, testSyncClaim), wrapTestWithProvisionCalls([]provisionCall{provision1Error}, testSyncClaim),
}, },
{ {
// Provision success - there is already a volume available, still // No provisioning if there is a matching volume available
// we provision a new one when requested.
"11-6 - provisioning when there is a volume available", "11-6 - provisioning when there is a volume available",
newVolumeArray("volume11-6", "1Gi", "", "", api.VolumePending, api.PersistentVolumeReclaimRetain), newVolumeArray("volume11-6", "1Gi", "", "", api.VolumePending, api.PersistentVolumeReclaimRetain, annClass),
[]*api.PersistentVolume{ newVolumeArray("volume11-6", "1Gi", "uid11-6", "claim11-6", api.VolumeBound, api.PersistentVolumeReclaimRetain, annBoundByController, annClass),
newVolume("volume11-6", "1Gi", "", "", api.VolumePending, api.PersistentVolumeReclaimRetain),
newVolume("pvc-uid11-6", "1Gi", "uid11-6", "claim11-6", api.VolumeBound, api.PersistentVolumeReclaimDelete, annBoundByController, annDynamicallyProvisioned),
},
newClaimArray("claim11-6", "uid11-6", "1Gi", "", api.ClaimPending, annClass),
// Binding will be completed in the next syncClaim
newClaimArray("claim11-6", "uid11-6", "1Gi", "", api.ClaimPending, annClass), newClaimArray("claim11-6", "uid11-6", "1Gi", "", api.ClaimPending, annClass),
newClaimArray("claim11-6", "uid11-6", "1Gi", "volume11-6", api.ClaimBound, annClass, annBoundByController, annBindCompleted),
noevents, noerrors, noevents, noerrors,
// No provisioning plugin confingure - makes the test fail when // No provisioning plugin confingure - makes the test fail when
// the controller errorneously tries to provision something // the controller errorneously tries to provision something
wrapTestWithControllerConfig(operationProvision, []error{nil}, testSyncClaim), wrapTestWithProvisionCalls([]provisionCall{provision1Success}, testSyncClaim),
}, },
{ {
// Provision success? - claim is bound before provisioner creates // Provision success? - claim is bound before provisioner creates
// a volume. // a volume.
"11-7 - claim is bound before provisioning", "11-7 - claim is bound before provisioning",
novolumes, novolumes,
newVolumeArray("pvc-uid11-7", "1Gi", "uid11-7", "claim11-7", api.VolumeBound, api.PersistentVolumeReclaimDelete, annBoundByController, annDynamicallyProvisioned), newVolumeArray("pvc-uid11-7", "1Gi", "uid11-7", "claim11-7", api.VolumeBound, api.PersistentVolumeReclaimDelete, annBoundByController, annDynamicallyProvisioned, annClass),
newClaimArray("claim11-7", "uid11-7", "1Gi", "", api.ClaimPending, annClass), newClaimArray("claim11-7", "uid11-7", "1Gi", "", api.ClaimPending, annClass),
// The claim would be bound in next syncClaim // The claim would be bound in next syncClaim
newClaimArray("claim11-7", "uid11-7", "1Gi", "", api.ClaimPending, annClass), newClaimArray("claim11-7", "uid11-7", "1Gi", "", api.ClaimPending, annClass),
noevents, noerrors, noevents, noerrors,
wrapTestWithInjectedOperation(wrapTestWithControllerConfig(operationProvision, []error{}, testSyncClaim), func(ctrl *PersistentVolumeController, reactor *volumeReactor) { wrapTestWithInjectedOperation(wrapTestWithProvisionCalls([]provisionCall{}, testSyncClaim), func(ctrl *PersistentVolumeController, reactor *volumeReactor) {
// Create a volume before provisionClaimOperation starts. // Create a volume before provisionClaimOperation starts.
// This similates a parallel controller provisioning the volume. // This similates a parallel controller provisioning the volume.
reactor.lock.Lock() reactor.lock.Lock()
volume := newVolume("pvc-uid11-7", "1Gi", "uid11-7", "claim11-7", api.VolumeBound, api.PersistentVolumeReclaimDelete, annBoundByController, annDynamicallyProvisioned) volume := newVolume("pvc-uid11-7", "1Gi", "uid11-7", "claim11-7", api.VolumeBound, api.PersistentVolumeReclaimDelete, annBoundByController, annDynamicallyProvisioned, annClass)
reactor.volumes[volume.Name] = volume reactor.volumes[volume.Name] = volume
reactor.lock.Unlock() reactor.lock.Unlock()
}), }),
@ -110,7 +160,7 @@ func TestProvisionSync(t *testing.T) {
// second retry succeeds // second retry succeeds
"11-8 - cannot save provisioned volume", "11-8 - cannot save provisioned volume",
novolumes, novolumes,
newVolumeArray("pvc-uid11-8", "1Gi", "uid11-8", "claim11-8", api.VolumeBound, api.PersistentVolumeReclaimDelete, annBoundByController, annDynamicallyProvisioned), newVolumeArray("pvc-uid11-8", "1Gi", "uid11-8", "claim11-8", api.VolumeBound, api.PersistentVolumeReclaimDelete, annBoundByController, annDynamicallyProvisioned, annClass),
newClaimArray("claim11-8", "uid11-8", "1Gi", "", api.ClaimPending, annClass), newClaimArray("claim11-8", "uid11-8", "1Gi", "", api.ClaimPending, annClass),
// Binding will be completed in the next syncClaim // Binding will be completed in the next syncClaim
newClaimArray("claim11-8", "uid11-8", "1Gi", "", api.ClaimPending, annClass), newClaimArray("claim11-8", "uid11-8", "1Gi", "", api.ClaimPending, annClass),
@ -121,7 +171,7 @@ func TestProvisionSync(t *testing.T) {
// will succeed. // will succeed.
{"create", "persistentvolumes", errors.New("Mock creation error")}, {"create", "persistentvolumes", errors.New("Mock creation error")},
}, },
wrapTestWithControllerConfig(operationProvision, []error{nil}, testSyncClaim), wrapTestWithProvisionCalls([]provisionCall{provision1Success}, testSyncClaim),
}, },
{ {
// Provision success? - cannot save provisioned PV five times, // Provision success? - cannot save provisioned PV five times,
@ -141,8 +191,12 @@ func TestProvisionSync(t *testing.T) {
{"create", "persistentvolumes", errors.New("Mock creation error4")}, {"create", "persistentvolumes", errors.New("Mock creation error4")},
{"create", "persistentvolumes", errors.New("Mock creation error5")}, {"create", "persistentvolumes", errors.New("Mock creation error5")},
}, },
wrapTestWithControllerConfig(operationDelete, []error{nil}, wrapTestWithPluginCalls(
wrapTestWithControllerConfig(operationProvision, []error{nil}, testSyncClaim)), nil, // recycle calls
[]error{nil}, // delete calls
[]provisionCall{provision1Success}, // provision calls
testSyncClaim,
),
}, },
{ {
// Provision failure - cannot save provisioned PV five times, // Provision failure - cannot save provisioned PV five times,
@ -163,7 +217,7 @@ func TestProvisionSync(t *testing.T) {
{"create", "persistentvolumes", errors.New("Mock creation error5")}, {"create", "persistentvolumes", errors.New("Mock creation error5")},
}, },
// No deleteCalls are configured, which results into no deleter plugin available for the volume // No deleteCalls are configured, which results into no deleter plugin available for the volume
wrapTestWithControllerConfig(operationProvision, []error{nil}, testSyncClaim), wrapTestWithProvisionCalls([]provisionCall{provision1Success}, testSyncClaim),
}, },
{ {
// Provision failure - cannot save provisioned PV five times, // Provision failure - cannot save provisioned PV five times,
@ -183,16 +237,17 @@ func TestProvisionSync(t *testing.T) {
{"create", "persistentvolumes", errors.New("Mock creation error4")}, {"create", "persistentvolumes", errors.New("Mock creation error4")},
{"create", "persistentvolumes", errors.New("Mock creation error5")}, {"create", "persistentvolumes", errors.New("Mock creation error5")},
}, },
wrapTestWithControllerConfig( wrapTestWithPluginCalls(
operationDelete, []error{ nil, // recycle calls
[]error{ // delete calls
errors.New("Mock deletion error1"), errors.New("Mock deletion error1"),
errors.New("Mock deletion error2"), errors.New("Mock deletion error2"),
errors.New("Mock deletion error3"), errors.New("Mock deletion error3"),
errors.New("Mock deletion error4"), errors.New("Mock deletion error4"),
errors.New("Mock deletion error5"), errors.New("Mock deletion error5"),
}, },
wrapTestWithControllerConfig(operationProvision, []error{nil}, testSyncClaim), []provisionCall{provision1Success}, // provision calls
), testSyncClaim),
}, },
{ {
// Provision failure - cannot save provisioned PV five times, // Provision failure - cannot save provisioned PV five times,
@ -212,16 +267,85 @@ func TestProvisionSync(t *testing.T) {
{"create", "persistentvolumes", errors.New("Mock creation error4")}, {"create", "persistentvolumes", errors.New("Mock creation error4")},
{"create", "persistentvolumes", errors.New("Mock creation error5")}, {"create", "persistentvolumes", errors.New("Mock creation error5")},
}, },
wrapTestWithControllerConfig( wrapTestWithPluginCalls(
operationDelete, []error{ nil, // recycle calls
[]error{ // delete calls
errors.New("Mock deletion error1"), errors.New("Mock deletion error1"),
nil, nil,
}, }, // provison calls
wrapTestWithControllerConfig(operationProvision, []error{nil}, testSyncClaim), []provisionCall{provision1Success},
testSyncClaim,
), ),
}, },
{
// Provision a volume (with non-default class)
"11-13 - successful provision with storage class 2",
novolumes,
volumeWithClass("silver", newVolumeArray("pvc-uid11-13", "1Gi", "uid11-13", "claim11-13", api.VolumeBound, api.PersistentVolumeReclaimDelete, annBoundByController, annDynamicallyProvisioned)),
claimWithClass("silver", newClaimArray("claim11-13", "uid11-13", "1Gi", "", api.ClaimPending)),
// Binding will be completed in the next syncClaim
claimWithClass("silver", newClaimArray("claim11-13", "uid11-13", "1Gi", "", api.ClaimPending)),
noevents, noerrors, wrapTestWithProvisionCalls([]provisionCall{provision2Success}, testSyncClaim),
},
{
// Provision error - non existing class
"11-14 - fail due to non-existing class",
novolumes,
novolumes,
claimWithClass("non-existing", newClaimArray("claim11-14", "uid11-14", "1Gi", "", api.ClaimPending)),
claimWithClass("non-existing", newClaimArray("claim11-14", "uid11-14", "1Gi", "", api.ClaimPending)),
noevents, noerrors, wrapTestWithProvisionCalls([]provisionCall{}, testSyncClaim),
},
{
// No provisioning with class=""
"11-15 - no provisioning with class=''",
novolumes,
novolumes,
claimWithClass("", newClaimArray("claim11-15", "uid11-15", "1Gi", "", api.ClaimPending)),
claimWithClass("", newClaimArray("claim11-15", "uid11-15", "1Gi", "", api.ClaimPending)),
noevents, noerrors, wrapTestWithProvisionCalls([]provisionCall{}, testSyncClaim),
},
{
// No provisioning with class=nil
"11-16 - no provisioning with class=nil",
novolumes,
novolumes,
newClaimArray("claim11-15", "uid11-15", "1Gi", "", api.ClaimPending),
newClaimArray("claim11-15", "uid11-15", "1Gi", "", api.ClaimPending),
noevents, noerrors, wrapTestWithProvisionCalls([]provisionCall{}, testSyncClaim),
},
} }
runSyncTests(t, tests) runSyncTests(t, tests, storageClasses)
}
func TestAlphaProvisionSync(t *testing.T) {
tests := []controllerTest{
{
// Provision a volume with alpha annotation
"14-1 - successful alpha provisioning",
novolumes,
newVolumeArray("pvc-uid14-1", "1Gi", "uid14-1", "claim14-1", api.VolumeBound, api.PersistentVolumeReclaimDelete, annBoundByController, annDynamicallyProvisioned),
newClaimArray("claim14-1", "uid14-1", "1Gi", "", api.ClaimPending, annAlphaClass),
// Binding will be completed in the next syncClaim
newClaimArray("claim14-1", "uid14-1", "1Gi", "", api.ClaimPending, annAlphaClass),
noevents, noerrors, wrapTestWithProvisionCalls([]provisionCall{provisionAlphaSuccess}, testSyncClaim),
},
{
// Provision success - there is already a volume available, still
// we provision a new one when requested.
"14-2 - no alpha provisioning when there is a volume available",
newVolumeArray("volume14-2", "1Gi", "", "", api.VolumePending, api.PersistentVolumeReclaimRetain),
[]*api.PersistentVolume{
newVolume("volume14-2", "1Gi", "", "", api.VolumePending, api.PersistentVolumeReclaimRetain),
newVolume("pvc-uid14-2", "1Gi", "uid14-2", "claim14-2", api.VolumeBound, api.PersistentVolumeReclaimDelete, annBoundByController, annDynamicallyProvisioned),
},
newClaimArray("claim14-2", "uid14-2", "1Gi", "", api.ClaimPending, annAlphaClass),
// Binding will be completed in the next syncClaim
newClaimArray("claim14-2", "uid14-2", "1Gi", "", api.ClaimPending, annAlphaClass),
noevents, noerrors, wrapTestWithProvisionCalls([]provisionCall{provisionAlphaSuccess}, testSyncClaim),
},
}
runSyncTests(t, tests, []*extensions.StorageClass{})
} }
// Test multiple calls to syncClaim/syncVolume and periodic sync of all // Test multiple calls to syncClaim/syncVolume and periodic sync of all
@ -244,20 +368,20 @@ func TestProvisionMultiSync(t *testing.T) {
// Provision a volume with binding // Provision a volume with binding
"12-1 - successful provision", "12-1 - successful provision",
novolumes, novolumes,
newVolumeArray("pvc-uid12-1", "1Gi", "uid12-1", "claim12-1", api.VolumeBound, api.PersistentVolumeReclaimDelete, annBoundByController, annDynamicallyProvisioned), newVolumeArray("pvc-uid12-1", "1Gi", "uid12-1", "claim12-1", api.VolumeBound, api.PersistentVolumeReclaimDelete, annBoundByController, annDynamicallyProvisioned, annClass),
newClaimArray("claim12-1", "uid12-1", "1Gi", "", api.ClaimPending, annClass), newClaimArray("claim12-1", "uid12-1", "1Gi", "", api.ClaimPending, annClass),
// Binding will be completed in the next syncClaim // Binding will be completed in the next syncClaim
newClaimArray("claim12-1", "uid12-1", "1Gi", "pvc-uid12-1", api.ClaimBound, annClass, annBoundByController, annBindCompleted), newClaimArray("claim12-1", "uid12-1", "1Gi", "pvc-uid12-1", api.ClaimBound, annClass, annBoundByController, annBindCompleted),
noevents, noerrors, wrapTestWithControllerConfig(operationProvision, []error{nil}, testSyncClaim), noevents, noerrors, wrapTestWithProvisionCalls([]provisionCall{provision1Success}, testSyncClaim),
}, },
} }
runMultisyncTests(t, tests) runMultisyncTests(t, tests, storageClasses, storageClasses[0].Name)
} }
// When provisioning is disabled, provisioning a claim should instantly return nil // When provisioning is disabled, provisioning a claim should instantly return nil
func TestDisablingDynamicProvisioner(t *testing.T) { func TestDisablingDynamicProvisioner(t *testing.T) {
ctrl := newTestController(nil, nil, nil, false) ctrl := newTestController(nil, nil, nil, nil, false)
retVal := ctrl.provisionClaim(nil) retVal := ctrl.provisionClaim(nil)
if retVal != nil { if retVal != nil {
t.Errorf("Expected nil return but got %v", retVal) t.Errorf("Expected nil return but got %v", retVal)

23
pkg/controller/volume/persistentvolume/recycle_test.go
View File

@ -21,6 +21,7 @@ import (
"testing" "testing"
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/apis/extensions"
) )
// Test single call to syncVolume, expecting recycling to happen. // Test single call to syncVolume, expecting recycling to happen.
@ -39,7 +40,7 @@ func TestRecycleSync(t *testing.T) {
noevents, noerrors, noevents, noerrors,
// Inject recycler into the controller and call syncVolume. The // Inject recycler into the controller and call syncVolume. The
// recycler simulates one recycle() call that succeeds. // recycler simulates one recycle() call that succeeds.
wrapTestWithControllerConfig(operationRecycle, []error{nil}, testSyncVolume), wrapTestWithReclaimCalls(operationRecycle, []error{nil}, testSyncVolume),
}, },
{ {
// recycle volume bound by user // recycle volume bound by user
@ -51,7 +52,7 @@ func TestRecycleSync(t *testing.T) {
noevents, noerrors, noevents, noerrors,
// Inject recycler into the controller and call syncVolume. The // Inject recycler into the controller and call syncVolume. The
// recycler simulates one recycle() call that succeeds. // recycler simulates one recycle() call that succeeds.
wrapTestWithControllerConfig(operationRecycle, []error{nil}, testSyncVolume), wrapTestWithReclaimCalls(operationRecycle, []error{nil}, testSyncVolume),
}, },
{ {
// recycle failure - plugin not found // recycle failure - plugin not found
@ -70,7 +71,7 @@ func TestRecycleSync(t *testing.T) {
noclaims, noclaims,
noclaims, noclaims,
[]string{"Warning VolumeFailedRecycle"}, noerrors, []string{"Warning VolumeFailedRecycle"}, noerrors,
wrapTestWithControllerConfig(operationRecycle, []error{}, testSyncVolume), wrapTestWithReclaimCalls(operationRecycle, []error{}, testSyncVolume),
}, },
{ {
// recycle failure - recycle returns error // recycle failure - recycle returns error
@ -80,7 +81,7 @@ func TestRecycleSync(t *testing.T) {
noclaims, noclaims,
noclaims, noclaims,
[]string{"Warning VolumeFailedRecycle"}, noerrors, []string{"Warning VolumeFailedRecycle"}, noerrors,
wrapTestWithControllerConfig(operationRecycle, []error{errors.New("Mock recycle error")}, testSyncVolume), wrapTestWithReclaimCalls(operationRecycle, []error{errors.New("Mock recycle error")}, testSyncVolume),
}, },
{ {
// recycle success(?) - volume is deleted before doRecycle() starts // recycle success(?) - volume is deleted before doRecycle() starts
@ -90,7 +91,7 @@ func TestRecycleSync(t *testing.T) {
noclaims, noclaims,
noclaims, noclaims,
noevents, noerrors, noevents, noerrors,
wrapTestWithInjectedOperation(wrapTestWithControllerConfig(operationRecycle, []error{}, testSyncVolume), func(ctrl *PersistentVolumeController, reactor *volumeReactor) { wrapTestWithInjectedOperation(wrapTestWithReclaimCalls(operationRecycle, []error{}, testSyncVolume), func(ctrl *PersistentVolumeController, reactor *volumeReactor) {
// Delete the volume before recycle operation starts // Delete the volume before recycle operation starts
reactor.lock.Lock() reactor.lock.Lock()
delete(reactor.volumes, "volume6-6") delete(reactor.volumes, "volume6-6")
@ -107,7 +108,7 @@ func TestRecycleSync(t *testing.T) {
noclaims, noclaims,
noclaims, noclaims,
noevents, noerrors, noevents, noerrors,
wrapTestWithInjectedOperation(wrapTestWithControllerConfig(operationRecycle, []error{}, testSyncVolume), func(ctrl *PersistentVolumeController, reactor *volumeReactor) { wrapTestWithInjectedOperation(wrapTestWithReclaimCalls(operationRecycle, []error{}, testSyncVolume), func(ctrl *PersistentVolumeController, reactor *volumeReactor) {
// Mark the volume as Available before the recycler starts // Mark the volume as Available before the recycler starts
reactor.lock.Lock() reactor.lock.Lock()
volume := reactor.volumes["volume6-7"] volume := reactor.volumes["volume6-7"]
@ -128,7 +129,7 @@ func TestRecycleSync(t *testing.T) {
noclaims, noclaims,
noclaims, noclaims,
noevents, noerrors, noevents, noerrors,
wrapTestWithInjectedOperation(wrapTestWithControllerConfig(operationRecycle, []error{}, testSyncVolume), func(ctrl *PersistentVolumeController, reactor *volumeReactor) { wrapTestWithInjectedOperation(wrapTestWithReclaimCalls(operationRecycle, []error{}, testSyncVolume), func(ctrl *PersistentVolumeController, reactor *volumeReactor) {
// Mark the volume as Available before the recycler starts // Mark the volume as Available before the recycler starts
reactor.lock.Lock() reactor.lock.Lock()
volume := reactor.volumes["volume6-8"] volume := reactor.volumes["volume6-8"]
@ -148,7 +149,7 @@ func TestRecycleSync(t *testing.T) {
noevents, noerrors, noevents, noerrors,
// Inject recycler into the controller and call syncVolume. The // Inject recycler into the controller and call syncVolume. The
// recycler simulates one recycle() call that succeeds. // recycler simulates one recycle() call that succeeds.
wrapTestWithControllerConfig(operationRecycle, []error{nil}, testSyncVolume), wrapTestWithReclaimCalls(operationRecycle, []error{nil}, testSyncVolume),
}, },
{ {
// volume has unknown reclaim policy - failure expected // volume has unknown reclaim policy - failure expected
@ -160,7 +161,7 @@ func TestRecycleSync(t *testing.T) {
[]string{"Warning VolumeUnknownReclaimPolicy"}, noerrors, testSyncVolume, []string{"Warning VolumeUnknownReclaimPolicy"}, noerrors, testSyncVolume,
}, },
} }
runSyncTests(t, tests) runSyncTests(t, tests, []*extensions.StorageClass{})
} }
// Test multiple calls to syncClaim/syncVolume and periodic sync of all // Test multiple calls to syncClaim/syncVolume and periodic sync of all
@ -188,9 +189,9 @@ func TestRecycleMultiSync(t *testing.T) {
noclaims, noclaims,
noclaims, noclaims,
[]string{"Warning VolumeFailedRecycle"}, noerrors, []string{"Warning VolumeFailedRecycle"}, noerrors,
wrapTestWithControllerConfig(operationRecycle, []error{errors.New("Mock recycle error"), nil}, testSyncVolume), wrapTestWithReclaimCalls(operationRecycle, []error{errors.New("Mock recycle error"), nil}, testSyncVolume),
}, },
} }
runMultisyncTests(t, tests) runMultisyncTests(t, tests, []*extensions.StorageClass{}, "")
} }

29
pkg/kubectl/resource_printer.go
View File

@ -491,6 +491,7 @@ var roleColumns = []string{"NAME", "AGE"}
var roleBindingColumns = []string{"NAME", "AGE"} var roleBindingColumns = []string{"NAME", "AGE"}
var clusterRoleColumns = []string{"NAME", "AGE"} var clusterRoleColumns = []string{"NAME", "AGE"}
var clusterRoleBindingColumns = []string{"NAME", "AGE"} var clusterRoleBindingColumns = []string{"NAME", "AGE"}
var storageClassColumns = []string{"NAME", "TYPE"}
// TODO: consider having 'KIND' for third party resource data // TODO: consider having 'KIND' for third party resource data
var thirdPartyResourceDataColumns = []string{"NAME", "LABELS", "DATA"} var thirdPartyResourceDataColumns = []string{"NAME", "LABELS", "DATA"}
@ -603,6 +604,8 @@ func (h *HumanReadablePrinter) addDefaultHandlers() {
h.Handler(clusterRoleBindingColumns, printClusterRoleBindingList) h.Handler(clusterRoleBindingColumns, printClusterRoleBindingList)
h.Handler(certificateSigningRequestColumns, printCertificateSigningRequest) h.Handler(certificateSigningRequestColumns, printCertificateSigningRequest)
h.Handler(certificateSigningRequestColumns, printCertificateSigningRequestList) h.Handler(certificateSigningRequestColumns, printCertificateSigningRequestList)
h.Handler(storageClassColumns, printStorageClass)
h.Handler(storageClassColumns, printStorageClassList)
} }
func (h *HumanReadablePrinter) unknown(data []byte, w io.Writer) error { func (h *HumanReadablePrinter) unknown(data []byte, w io.Writer) error {
@ -2067,6 +2070,32 @@ func printNetworkPolicyList(list *extensions.NetworkPolicyList, w io.Writer, opt
return nil return nil
} }
func printStorageClass(sc *extensions.StorageClass, w io.Writer, options PrintOptions) error {
name := sc.Name
provtype := sc.Provisioner
if _, err := fmt.Fprintf(w, "%s\t%s\t", name, provtype); err != nil {
return err
}
if _, err := fmt.Fprint(w, AppendLabels(sc.Labels, options.ColumnLabels)); err != nil {
return err
}
if _, err := fmt.Fprint(w, AppendAllLabels(options.ShowLabels, sc.Labels)); err != nil {
return err
}
return nil
}
func printStorageClassList(scList *extensions.StorageClassList, w io.Writer, options PrintOptions) error {
for _, sc := range scList.Items {
if err := printStorageClass(&sc, w, options); err != nil {
return err
}
}
return nil
}
func AppendLabels(itemLabels map[string]string, columnLabels []string) string { func AppendLabels(itemLabels map[string]string, columnLabels []string) string {
var buffer bytes.Buffer var buffer bytes.Buffer

24
pkg/volume/aws_ebs/aws_util.go
View File

@ -18,6 +18,8 @@ package aws_ebs
import ( import (
"fmt" "fmt"
"strconv"
"strings"
"time" "time"
"github.com/golang/glog" "github.com/golang/glog"
@ -83,6 +85,28 @@ func (util *AWSDiskUtil) CreateVolume(c *awsElasticBlockStoreProvisioner) (strin
Tags: tags, Tags: tags,
PVCName: c.options.PVCName, PVCName: c.options.PVCName,
} }
// Apply Parameters (case-insensitive). We leave validation of
// the values to the cloud provider.
for k, v := range c.options.Parameters {
switch strings.ToLower(k) {
case "type":
volumeOptions.VolumeType = v
case "zone":
volumeOptions.AvailabilityZone = v
case "iopspergb":
volumeOptions.IOPSPerGB, err = strconv.Atoi(v)
if err != nil {
return "", 0, nil, fmt.Errorf("invalid iopsPerGB value %q, must be integer between 1 and 30: %v", v, err)
}
default:
return "", 0, nil, fmt.Errorf("invalid option %q for volume plugin %s", k, c.plugin.GetPluginName())
}
}
// TODO: implement c.options.ProvisionerSelector parsing
if c.options.Selector != nil {
return "", 0, nil, fmt.Errorf("claim.Spec.Selector is not supported for dynamic provisioning on AWS")
}
name, err := cloud.CreateDisk(volumeOptions) name, err := cloud.CreateDisk(volumeOptions)
if err != nil { if err != nil {

2
pkg/volume/gce_pd/attacher_test.go
View File

@ -351,7 +351,7 @@ func (testcase *testcase) DiskIsAttached(diskName, instanceID string) (bool, err
return expected.isAttached, expected.ret return expected.isAttached, expected.ret
} }
func (testcase *testcase) CreateDisk(name string, zone string, sizeGb int64, tags map[string]string) error { func (testcase *testcase) CreateDisk(name string, diskType string, zone string, sizeGb int64, tags map[string]string) error {
return errors.New("Not implemented") return errors.New("Not implemented")
} }

37
pkg/volume/gce_pd/gce_util.go
View File

@ -79,17 +79,38 @@ func (gceutil *GCEDiskUtil) CreateVolume(c *gcePersistentDiskProvisioner) (strin
// GCE works with gigabytes, convert to GiB with rounding up // GCE works with gigabytes, convert to GiB with rounding up
requestGB := volume.RoundUpSize(requestBytes, 1024*1024*1024) requestGB := volume.RoundUpSize(requestBytes, 1024*1024*1024)
// The disk will be created in the zone in which this code is currently running // Apply Parameters (case-insensitive). We leave validation of
// TODO: We should support auto-provisioning volumes in multiple/specified zones // the values to the cloud provider.
zones, err := cloud.GetAllZones() diskType := ""
if err != nil { zone := ""
glog.V(2).Infof("error getting zone information from GCE: %v", err) for k, v := range c.options.Parameters {
return "", 0, nil, err switch strings.ToLower(k) {
case "type":
diskType = v
case "zone":
zone = v
default:
return "", 0, nil, fmt.Errorf("invalid option %q for volume plugin %s", k, c.plugin.GetPluginName())
}
} }
zone := volume.ChooseZoneForVolume(zones, c.options.PVCName) // TODO: implement c.options.ProvisionerSelector parsing
if c.options.Selector != nil {
return "", 0, nil, fmt.Errorf("claim.Spec.Selector is not supported for dynamic provisioning on GCE")
}
err = cloud.CreateDisk(name, zone, int64(requestGB), *c.options.CloudTags) if zone == "" {
// No zone specified, choose one randomly in the same region as the
// node is running.
zones, err := cloud.GetAllZones()
if err != nil {
glog.V(2).Infof("error getting zone information from GCE: %v", err)
return "", 0, nil, err
}
zone = volume.ChooseZoneForVolume(zones, c.options.PVCName)
}
err = cloud.CreateDisk(name, diskType, zone, int64(requestGB), *c.options.CloudTags)
if err != nil { if err != nil {
glog.V(2).Infof("Error creating GCE PD volume: %v", err) glog.V(2).Infof("Error creating GCE PD volume: %v", err)
return "", 0, nil, err return "", 0, nil, err

14
pkg/volume/host_path/host_path.go
View File

@ -43,17 +43,6 @@ func ProbeVolumePlugins(volumeConfig volume.VolumeConfig) []volume.VolumePlugin
} }
} }
func ProbeRecyclableVolumePlugins(recyclerFunc func(pvName string, spec *volume.Spec, host volume.VolumeHost, volumeConfig volume.VolumeConfig) (volume.Recycler, error), volumeConfig volume.VolumeConfig) []volume.VolumePlugin {
return []volume.VolumePlugin{
&hostPathPlugin{
host: nil,
newRecyclerFunc: recyclerFunc,
newProvisionerFunc: newProvisioner,
config: volumeConfig,
},
}
}
type hostPathPlugin struct { type hostPathPlugin struct {
host volume.VolumeHost host volume.VolumeHost
// decouple creating Recyclers/Deleters/Provisioners by deferring to a function. Allows for easier testing. // decouple creating Recyclers/Deleters/Provisioners by deferring to a function. Allows for easier testing.
@ -132,6 +121,9 @@ func (plugin *hostPathPlugin) NewDeleter(spec *volume.Spec) (volume.Deleter, err
} }
func (plugin *hostPathPlugin) NewProvisioner(options volume.VolumeOptions) (volume.Provisioner, error) { func (plugin *hostPathPlugin) NewProvisioner(options volume.VolumeOptions) (volume.Provisioner, error) {
if !plugin.config.ProvisioningEnabled {
return nil, fmt.Errorf("Provisioning in volume plugin %q is disabled", plugin.GetPluginName())
}
if len(options.AccessModes) == 0 { if len(options.AccessModes) == 0 {
options.AccessModes = plugin.GetAccessModes() options.AccessModes = plugin.GetAccessModes()
} }

3
pkg/volume/host_path/host_path_test.go
View File

@ -154,7 +154,8 @@ func TestProvisioner(t *testing.T) {
err := os.MkdirAll(tempPath, 0750) err := os.MkdirAll(tempPath, 0750)
plugMgr := volume.VolumePluginMgr{} plugMgr := volume.VolumePluginMgr{}
plugMgr.InitPlugins(ProbeVolumePlugins(volume.VolumeConfig{}), volumetest.NewFakeVolumeHost("/tmp/fake", nil, nil, "" /* rootContext */)) plugMgr.InitPlugins(ProbeVolumePlugins(volume.VolumeConfig{ProvisioningEnabled: true}),
volumetest.NewFakeVolumeHost("/tmp/fake", nil, nil, "" /* rootContext */))
spec := &volume.Spec{PersistentVolume: &api.PersistentVolume{Spec: api.PersistentVolumeSpec{PersistentVolumeSource: api.PersistentVolumeSource{HostPath: &api.HostPathVolumeSource{Path: tempPath}}}}} spec := &volume.Spec{PersistentVolume: &api.PersistentVolume{Spec: api.PersistentVolumeSpec{PersistentVolumeSource: api.PersistentVolumeSource{HostPath: &api.HostPathVolumeSource{Path: tempPath}}}}}
plug, err := plugMgr.FindCreatablePluginBySpec(spec) plug, err := plugMgr.FindCreatablePluginBySpec(spec)
if err != nil { if err != nil {

37
pkg/volume/plugins.go
View File

@ -25,6 +25,7 @@ import (
"github.com/golang/glog" "github.com/golang/glog"
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/resource" "k8s.io/kubernetes/pkg/api/resource"
"k8s.io/kubernetes/pkg/api/unversioned"
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset" clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
"k8s.io/kubernetes/pkg/cloudprovider" "k8s.io/kubernetes/pkg/cloudprovider"
"k8s.io/kubernetes/pkg/types" "k8s.io/kubernetes/pkg/types"
@ -55,6 +56,10 @@ type VolumeOptions struct {
ClusterName string ClusterName string
// Tags to attach to the real volume in the cloud provider - e.g. AWS EBS // Tags to attach to the real volume in the cloud provider - e.g. AWS EBS
CloudTags *map[string]string CloudTags *map[string]string
// Volume provisioning parameters from StorageClass
Parameters map[string]string
// Volume selector from PersistentVolumeClaim
Selector *unversioned.LabelSelector
} }
// VolumePlugin is an interface to volume plugins that can be used on a // VolumePlugin is an interface to volume plugins that can be used on a
@ -293,6 +298,10 @@ type VolumeConfig struct {
// the system and only understood by the binary hosting the plugin and the // the system and only understood by the binary hosting the plugin and the
// plugin itself. // plugin itself.
OtherAttributes map[string]string OtherAttributes map[string]string
// ProvisioningEnabled configures whether provisioning of this plugin is
// enabled or not. Currently used only in host_path plugin.
ProvisioningEnabled bool
} }
// NewSpecFromVolume creates an Spec from an api.Volume // NewSpecFromVolume creates an Spec from an api.Volume
@ -429,7 +438,20 @@ func (pm *VolumePluginMgr) FindRecyclablePluginBySpec(spec *Spec) (RecyclableVol
return nil, fmt.Errorf("no recyclable volume plugin matched") return nil, fmt.Errorf("no recyclable volume plugin matched")
} }
// FindDeletablePluginByName fetches a persistent volume plugin by name. If // FindProvisionablePluginByName fetches a persistent volume plugin by name. If
// no plugin is found, returns error.
func (pm *VolumePluginMgr) FindProvisionablePluginByName(name string) (ProvisionableVolumePlugin, error) {
volumePlugin, err := pm.FindPluginByName(name)
if err != nil {
return nil, err
}
if provisionableVolumePlugin, ok := volumePlugin.(ProvisionableVolumePlugin); ok {
return provisionableVolumePlugin, nil
}
return nil, fmt.Errorf("no provisionable volume plugin matched")
}
// FindDeletablePluginBySppec fetches a persistent volume plugin by spec. If
// no plugin is found, returns error. // no plugin is found, returns error.
func (pm *VolumePluginMgr) FindDeletablePluginBySpec(spec *Spec) (DeletableVolumePlugin, error) { func (pm *VolumePluginMgr) FindDeletablePluginBySpec(spec *Spec) (DeletableVolumePlugin, error) {
volumePlugin, err := pm.FindPluginBySpec(spec) volumePlugin, err := pm.FindPluginBySpec(spec)
@ -442,6 +464,19 @@ func (pm *VolumePluginMgr) FindDeletablePluginBySpec(spec *Spec) (DeletableVolum
return nil, fmt.Errorf("no deletable volume plugin matched") return nil, fmt.Errorf("no deletable volume plugin matched")
} }
// FindDeletablePluginByName fetches a persistent volume plugin by name. If
// no plugin is found, returns error.
func (pm *VolumePluginMgr) FindDeletablePluginByName(name string) (DeletableVolumePlugin, error) {
volumePlugin, err := pm.FindPluginByName(name)
if err != nil {
return nil, err
}
if deletableVolumePlugin, ok := volumePlugin.(DeletableVolumePlugin); ok {
return deletableVolumePlugin, nil
}
return nil, fmt.Errorf("no deletable volume plugin matched")
}
// FindCreatablePluginBySpec fetches a persistent volume plugin by name. If // FindCreatablePluginBySpec fetches a persistent volume plugin by name. If
// no plugin is found, returns error. // no plugin is found, returns error.
func (pm *VolumePluginMgr) FindCreatablePluginBySpec(spec *Spec) (ProvisionableVolumePlugin, error) { func (pm *VolumePluginMgr) FindCreatablePluginBySpec(spec *Spec) (ProvisionableVolumePlugin, error) {

2
test/e2e/pd.go
View File

@ -469,7 +469,7 @@ func createPD() (string, error) {
} }
tags := map[string]string{} tags := map[string]string{}
err = gceCloud.CreateDisk(pdName, framework.TestContext.CloudConfig.Zone, 10 /* sizeGb */, tags) err = gceCloud.CreateDisk(pdName, gcecloud.DiskTypeSSD, framework.TestContext.CloudConfig.Zone, 10 /* sizeGb */, tags)
if err != nil { if err != nil {
return "", err return "", err
} }

193
test/e2e/volume_provisioning.go
View File

@ -22,6 +22,7 @@ import (
"k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/resource" "k8s.io/kubernetes/pkg/api/resource"
"k8s.io/kubernetes/pkg/api/unversioned" "k8s.io/kubernetes/pkg/api/unversioned"
"k8s.io/kubernetes/pkg/apis/extensions"
client "k8s.io/kubernetes/pkg/client/unversioned" client "k8s.io/kubernetes/pkg/client/unversioned"
"k8s.io/kubernetes/test/e2e/framework" "k8s.io/kubernetes/test/e2e/framework"
@ -37,6 +38,69 @@ const (
expectedSize = "2Gi" expectedSize = "2Gi"
) )
func testDynamicProvisioning(client *client.Client, claim *api.PersistentVolumeClaim) {
err := framework.WaitForPersistentVolumeClaimPhase(api.ClaimBound, client, claim.Namespace, claim.Name, framework.Poll, framework.ClaimProvisionTimeout)
Expect(err).NotTo(HaveOccurred())
By("checking the claim")
// Get new copy of the claim
claim, err = client.PersistentVolumeClaims(claim.Namespace).Get(claim.Name)
Expect(err).NotTo(HaveOccurred())
// Get the bound PV
pv, err := client.PersistentVolumes().Get(claim.Spec.VolumeName)
Expect(err).NotTo(HaveOccurred())
// Check sizes
expectedCapacity := resource.MustParse(expectedSize)
pvCapacity := pv.Spec.Capacity[api.ResourceName(api.ResourceStorage)]
Expect(pvCapacity.Value()).To(Equal(expectedCapacity.Value()))
requestedCapacity := resource.MustParse(requestedSize)
claimCapacity := claim.Spec.Resources.Requests[api.ResourceName(api.ResourceStorage)]
Expect(claimCapacity.Value()).To(Equal(requestedCapacity.Value()))
// Check PV properties
Expect(pv.Spec.PersistentVolumeReclaimPolicy).To(Equal(api.PersistentVolumeReclaimDelete))
expectedAccessModes := []api.PersistentVolumeAccessMode{api.ReadWriteOnce}
Expect(pv.Spec.AccessModes).To(Equal(expectedAccessModes))
Expect(pv.Spec.ClaimRef.Name).To(Equal(claim.ObjectMeta.Name))
Expect(pv.Spec.ClaimRef.Namespace).To(Equal(claim.ObjectMeta.Namespace))
// We start two pods:
// - The first writes 'hello word' to the /mnt/test (= the volume).
// - The second one runs grep 'hello world' on /mnt/test.
// If both succeed, Kubernetes actually allocated something that is
// persistent across pods.
By("checking the created volume is writable")
runInPodWithVolume(client, claim.Namespace, claim.Name, "echo 'hello world' > /mnt/test/data")
By("checking the created volume is readable and retains data")
runInPodWithVolume(client, claim.Namespace, claim.Name, "grep 'hello world' /mnt/test/data")
// Ugly hack: if we delete the AWS/GCE/OpenStack volume here, it will
// probably collide with destruction of the pods above - the pods
// still have the volume attached (kubelet is slow...) and deletion
// of attached volume is not allowed by AWS/GCE/OpenStack.
// Kubernetes *will* retry deletion several times in
// pvclaimbinder-sync-period.
// So, technically, this sleep is not needed. On the other hand,
// the sync perion is 10 minutes and we really don't want to wait
// 10 minutes here. There is no way how to see if kubelet is
// finished with cleaning volumes. A small sleep here actually
// speeds up the test!
// Three minutes should be enough to clean up the pods properly.
// We've seen GCE PD detach to take more than 1 minute.
By("Sleeping to let kubelet destroy all pods")
time.Sleep(3 * time.Minute)
By("deleting the claim")
framework.ExpectNoError(client.PersistentVolumeClaims(claim.Namespace).Delete(claim.Name))
// Wait for the PV to get deleted too.
framework.ExpectNoError(framework.WaitForPersistentVolumeDeleted(client, pv.Name, 5*time.Second, 20*time.Minute))
}
var _ = framework.KubeDescribe("Dynamic provisioning", func() { var _ = framework.KubeDescribe("Dynamic provisioning", func() {
f := framework.NewDefaultFramework("volume-provisioning") f := framework.NewDefaultFramework("volume-provisioning")
@ -52,86 +116,47 @@ var _ = framework.KubeDescribe("Dynamic provisioning", func() {
framework.KubeDescribe("DynamicProvisioner", func() { framework.KubeDescribe("DynamicProvisioner", func() {
It("should create and delete persistent volumes", func() { It("should create and delete persistent volumes", func() {
framework.SkipUnlessProviderIs("openstack", "gce", "aws", "gke") framework.SkipUnlessProviderIs("openstack", "gce", "aws", "gke")
By("creating a StorageClass")
class := newStorageClass()
_, err := c.Extensions().StorageClasses().Create(class)
defer c.Extensions().StorageClasses().Delete(class.Name)
Expect(err).NotTo(HaveOccurred())
By("creating a claim with a dynamic provisioning annotation") By("creating a claim with a dynamic provisioning annotation")
claim := createClaim(ns) claim := newClaim(ns, false)
defer func() {
c.PersistentVolumeClaims(ns).Delete(claim.Name)
}()
claim, err = c.PersistentVolumeClaims(ns).Create(claim)
Expect(err).NotTo(HaveOccurred())
testDynamicProvisioning(c, claim)
})
})
framework.KubeDescribe("DynamicProvisioner Alpha", func() {
It("should create and delete alpha persistent volumes", func() {
framework.SkipUnlessProviderIs("openstack", "gce", "aws", "gke")
By("creating a claim with an alpha dynamic provisioning annotation")
claim := newClaim(ns, true)
defer func() { defer func() {
c.PersistentVolumeClaims(ns).Delete(claim.Name) c.PersistentVolumeClaims(ns).Delete(claim.Name)
}() }()
claim, err := c.PersistentVolumeClaims(ns).Create(claim) claim, err := c.PersistentVolumeClaims(ns).Create(claim)
Expect(err).NotTo(HaveOccurred()) Expect(err).NotTo(HaveOccurred())
err = framework.WaitForPersistentVolumeClaimPhase(api.ClaimBound, c, ns, claim.Name, framework.Poll, framework.ClaimProvisionTimeout) testDynamicProvisioning(c, claim)
Expect(err).NotTo(HaveOccurred())
By("checking the claim")
// Get new copy of the claim
claim, err = c.PersistentVolumeClaims(ns).Get(claim.Name)
Expect(err).NotTo(HaveOccurred())
// Get the bound PV
pv, err := c.PersistentVolumes().Get(claim.Spec.VolumeName)
Expect(err).NotTo(HaveOccurred())
// Check sizes
expectedCapacity := resource.MustParse(expectedSize)
pvCapacity := pv.Spec.Capacity[api.ResourceName(api.ResourceStorage)]
Expect(pvCapacity.Value()).To(Equal(expectedCapacity.Value()))
requestedCapacity := resource.MustParse(requestedSize)
claimCapacity := claim.Spec.Resources.Requests[api.ResourceName(api.ResourceStorage)]
Expect(claimCapacity.Value()).To(Equal(requestedCapacity.Value()))
// Check PV properties
Expect(pv.Spec.PersistentVolumeReclaimPolicy).To(Equal(api.PersistentVolumeReclaimDelete))
expectedAccessModes := []api.PersistentVolumeAccessMode{api.ReadWriteOnce}
Expect(pv.Spec.AccessModes).To(Equal(expectedAccessModes))
Expect(pv.Spec.ClaimRef.Name).To(Equal(claim.ObjectMeta.Name))
Expect(pv.Spec.ClaimRef.Namespace).To(Equal(claim.ObjectMeta.Namespace))
// We start two pods:
// - The first writes 'hello word' to the /mnt/test (= the volume).
// - The second one runs grep 'hello world' on /mnt/test.
// If both succeed, Kubernetes actually allocated something that is
// persistent across pods.
By("checking the created volume is writable")
runInPodWithVolume(c, ns, claim.Name, "echo 'hello world' > /mnt/test/data")
By("checking the created volume is readable and retains data")
runInPodWithVolume(c, ns, claim.Name, "grep 'hello world' /mnt/test/data")
// Ugly hack: if we delete the AWS/GCE/OpenStack volume here, it will
// probably collide with destruction of the pods above - the pods
// still have the volume attached (kubelet is slow...) and deletion
// of attached volume is not allowed by AWS/GCE/OpenStack.
// Kubernetes *will* retry deletion several times in
// pvclaimbinder-sync-period.
// So, technically, this sleep is not needed. On the other hand,
// the sync perion is 10 minutes and we really don't want to wait
// 10 minutes here. There is no way how to see if kubelet is
// finished with cleaning volumes. A small sleep here actually
// speeds up the test!
// Three minutes should be enough to clean up the pods properly.
// We've seen GCE PD detach to take more than 1 minute.
By("Sleeping to let kubelet destroy all pods")
time.Sleep(3 * time.Minute)
By("deleting the claim")
framework.ExpectNoError(c.PersistentVolumeClaims(ns).Delete(claim.Name))
// Wait for the PV to get deleted too.
framework.ExpectNoError(framework.WaitForPersistentVolumeDeleted(c, pv.Name, 5*time.Second, 20*time.Minute))
}) })
}) })
}) })
func createClaim(ns string) *api.PersistentVolumeClaim { func newClaim(ns string, alpha bool) *api.PersistentVolumeClaim {
return &api.PersistentVolumeClaim{ claim := api.PersistentVolumeClaim{
ObjectMeta: api.ObjectMeta{ ObjectMeta: api.ObjectMeta{
GenerateName: "pvc-", GenerateName: "pvc-",
Namespace: ns, Namespace: ns,
Annotations: map[string]string{
"volume.alpha.kubernetes.io/storage-class": "",
},
}, },
Spec: api.PersistentVolumeClaimSpec{ Spec: api.PersistentVolumeClaimSpec{
AccessModes: []api.PersistentVolumeAccessMode{ AccessModes: []api.PersistentVolumeAccessMode{
@ -144,6 +169,19 @@ func createClaim(ns string) *api.PersistentVolumeClaim {
}, },
}, },
} }
if alpha {
claim.Annotations = map[string]string{
"volume.alpha.kubernetes.io/storage-class": "",
}
} else {
claim.Annotations = map[string]string{
"volume.beta.kubernetes.io/storage-class": "fast",
}
}
return &claim
} }
// runInPodWithVolume runs a command in a pod with given claim mounted to /mnt directory. // runInPodWithVolume runs a command in a pod with given claim mounted to /mnt directory.
@ -192,3 +230,26 @@ func runInPodWithVolume(c *client.Client, ns, claimName, command string) {
framework.ExpectNoError(err, "Failed to create pod: %v", err) framework.ExpectNoError(err, "Failed to create pod: %v", err)
framework.ExpectNoError(framework.WaitForPodSuccessInNamespaceSlow(c, pod.Name, pod.Spec.Containers[0].Name, pod.Namespace)) framework.ExpectNoError(framework.WaitForPodSuccessInNamespaceSlow(c, pod.Name, pod.Spec.Containers[0].Name, pod.Namespace))
} }
func newStorageClass() *extensions.StorageClass {
var pluginName string
switch {
case framework.ProviderIs("gke"), framework.ProviderIs("gce"):
pluginName = "kubernetes.io/gce-pd"
case framework.ProviderIs("aws"):
pluginName = "kubernetes.io/aws-ebs"
case framework.ProviderIs("openstack"):
pluginName = "kubernetes.io/cinder"
}
return &extensions.StorageClass{
TypeMeta: unversioned.TypeMeta{
Kind: "StorageClass",
},
ObjectMeta: api.ObjectMeta{
Name: "fast",
},
Provisioner: pluginName,
}
}

34
test/integration/persistentvolumes/persistent_volumes_test.go
View File

@ -31,6 +31,7 @@ import (
"k8s.io/kubernetes/pkg/api/resource" "k8s.io/kubernetes/pkg/api/resource"
"k8s.io/kubernetes/pkg/api/testapi" "k8s.io/kubernetes/pkg/api/testapi"
"k8s.io/kubernetes/pkg/api/unversioned" "k8s.io/kubernetes/pkg/api/unversioned"
"k8s.io/kubernetes/pkg/apis/extensions"
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset" clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
"k8s.io/kubernetes/pkg/client/restclient" "k8s.io/kubernetes/pkg/client/restclient"
fake_cloud "k8s.io/kubernetes/pkg/cloudprovider/providers/fake" fake_cloud "k8s.io/kubernetes/pkg/cloudprovider/providers/fake"
@ -62,6 +63,8 @@ func init() {
const defaultObjectCount = 100 const defaultObjectCount = 100
const defaultSyncPeriod = 10 * time.Second const defaultSyncPeriod = 10 * time.Second
const provisionerPluginName = "kubernetes.io/mock-provisioner"
func getObjectCount() int { func getObjectCount() int {
objectCount := defaultObjectCount objectCount := defaultObjectCount
if s := os.Getenv("KUBE_INTEGRATION_PV_OBJECTS"); s != "" { if s := os.Getenv("KUBE_INTEGRATION_PV_OBJECTS"); s != "" {
@ -849,8 +852,20 @@ func TestPersistentVolumeProvisionMultiPVCs(t *testing.T) {
defer watchPVC.Stop() defer watchPVC.Stop()
// NOTE: This test cannot run in parallel, because it is creating and deleting // NOTE: This test cannot run in parallel, because it is creating and deleting
// non-namespaced objects (PersistenceVolumes). // non-namespaced objects (PersistenceVolumes and StorageClasses).
defer testClient.Core().PersistentVolumes().DeleteCollection(nil, api.ListOptions{}) defer testClient.Core().PersistentVolumes().DeleteCollection(nil, api.ListOptions{})
defer testClient.Extensions().StorageClasses().DeleteCollection(nil, api.ListOptions{})
storageClass := extensions.StorageClass{
TypeMeta: unversioned.TypeMeta{
Kind: "StorageClass",
},
ObjectMeta: api.ObjectMeta{
Name: "gold",
},
Provisioner: provisionerPluginName,
}
testClient.Extensions().StorageClasses().Create(&storageClass)
binder.Run() binder.Run()
defer binder.Stop() defer binder.Stop()
@ -860,7 +875,7 @@ func TestPersistentVolumeProvisionMultiPVCs(t *testing.T) {
for i := 0; i < objCount; i++ { for i := 0; i < objCount; i++ {
pvc := createPVC("pvc-provision-"+strconv.Itoa(i), ns.Name, "1G", []api.PersistentVolumeAccessMode{api.ReadWriteOnce}) pvc := createPVC("pvc-provision-"+strconv.Itoa(i), ns.Name, "1G", []api.PersistentVolumeAccessMode{api.ReadWriteOnce})
pvc.Annotations = map[string]string{ pvc.Annotations = map[string]string{
"volume.alpha.kubernetes.io/storage-class": "", "volume.beta.kubernetes.io/storage-class": "gold",
} }
pvcs[i] = pvc pvcs[i] = pvc
} }
@ -1086,7 +1101,7 @@ func createClients(ns *api.Namespace, t *testing.T, s *httptest.Server, syncPeri
host := volumetest.NewFakeVolumeHost("/tmp/fake", nil, nil, "" /* rootContext */) host := volumetest.NewFakeVolumeHost("/tmp/fake", nil, nil, "" /* rootContext */)
plugin := &volumetest.FakeVolumePlugin{ plugin := &volumetest.FakeVolumePlugin{
PluginName: "plugin-name", PluginName: provisionerPluginName,
Host: host, Host: host,
Config: volume.VolumeConfig{}, Config: volume.VolumeConfig{},
LastProvisionerOptions: volume.VolumeOptions{}, LastProvisionerOptions: volume.VolumeOptions{},
@ -1101,7 +1116,18 @@ func createClients(ns *api.Namespace, t *testing.T, s *httptest.Server, syncPeri
cloud := &fake_cloud.FakeCloud{} cloud := &fake_cloud.FakeCloud{}
syncPeriod = getSyncPeriod(syncPeriod) syncPeriod = getSyncPeriod(syncPeriod)
ctrl := persistentvolumecontroller.NewPersistentVolumeController(binderClient, syncPeriod, plugin, plugins, cloud, "", nil, nil, nil, true) ctrl := persistentvolumecontroller.NewPersistentVolumeController(
binderClient,
syncPeriod,
nil, // alpha provisioner
plugins,
cloud,
"", // cluster name
nil, // volumeSource
nil, // claimSource
nil, // classSource
nil, // eventRecorder
true) // enableDynamicProvisioning
watchPV, err := testClient.PersistentVolumes().Watch(api.ListOptions{}) watchPV, err := testClient.PersistentVolumes().Watch(api.ListOptions{})
if err != nil { if err != nil {