github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-27 21:47:07 +00:00
Code Issues Projects Releases Wiki Activity

refactor persistent volume labeler admission controller to use cloudprovider.PVLabler

Browse Source
This commit is contained in:
andrewsykim 2019-01-18 18:56:26 -05:00 committed by Andrew Kim
parent 20a11ac5cc
commit 1a316015e3
3 changed files with 76 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
plugin/pkg/admission/storage/persistentvolume/label/BUILD
View File

@ -15,13 +15,12 @@ go_library(
importpath = "k8s.io/kubernetes/plugin/pkg/admission/storage/persistentvolume/label", importpath = "k8s.io/kubernetes/plugin/pkg/admission/storage/persistentvolume/label",
deps = [ deps = [
"//pkg/apis/core:go_default_library", "//pkg/apis/core:go_default_library",
"//pkg/cloudprovider/providers/aws:go_default_library", "//pkg/apis/core/v1:go_default_library",
"//pkg/cloudprovider/providers/azure:go_default_library",
"//pkg/cloudprovider/providers/gce:go_default_library",
"//pkg/kubeapiserver/admission:go_default_library", "//pkg/kubeapiserver/admission:go_default_library",
"//pkg/kubelet/apis:go_default_library", "//pkg/kubelet/apis:go_default_library",
"//pkg/volume:go_default_library", "//pkg/volume:go_default_library",
"//pkg/volume/util:go_default_library", "//pkg/volume/util:go_default_library",
"//staging/src/k8s.io/api/core/v1:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/admission:go_default_library", "//staging/src/k8s.io/apiserver/pkg/admission:go_default_library",
"//staging/src/k8s.io/cloud-provider:go_default_library", "//staging/src/k8s.io/cloud-provider:go_default_library",
"//vendor/k8s.io/klog:go_default_library", "//vendor/k8s.io/klog:go_default_library",
@ -34,13 +33,12 @@ go_test(
embed = [":go_default_library"], embed = [":go_default_library"],
deps = [ deps = [
"//pkg/apis/core:go_default_library", "//pkg/apis/core:go_default_library",
"//pkg/cloudprovider/providers/aws:go_default_library",
"//pkg/kubelet/apis:go_default_library", "//pkg/kubelet/apis:go_default_library",
"//pkg/volume/util:go_default_library", "//pkg/volume/util:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/api/resource:go_default_library", "//staging/src/k8s.io/api/core/v1:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library", "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/admission:go_default_library", "//staging/src/k8s.io/apiserver/pkg/admission:go_default_library",
"//staging/src/k8s.io/cloud-provider:go_default_library",
], ],
) )

112
plugin/pkg/admission/storage/persistentvolume/label/admission.go
View File

@ -18,17 +18,18 @@ package label
import ( import (
"bytes" "bytes"
"context"
"errors"
"fmt" "fmt"
"io" "io"
"sync" "sync"
v1 "k8s.io/api/core/v1"
"k8s.io/apiserver/pkg/admission" "k8s.io/apiserver/pkg/admission"
cloudprovider "k8s.io/cloud-provider" cloudprovider "k8s.io/cloud-provider"
"k8s.io/klog" "k8s.io/klog"
api "k8s.io/kubernetes/pkg/apis/core" api "k8s.io/kubernetes/pkg/apis/core"
"k8s.io/kubernetes/pkg/cloudprovider/providers/aws" k8s_api_v1 "k8s.io/kubernetes/pkg/apis/core/v1"
"k8s.io/kubernetes/pkg/cloudprovider/providers/azure"
"k8s.io/kubernetes/pkg/cloudprovider/providers/gce"
kubeapiserveradmission "k8s.io/kubernetes/pkg/kubeapiserver/admission" kubeapiserveradmission "k8s.io/kubernetes/pkg/kubeapiserver/admission"
kubeletapis "k8s.io/kubernetes/pkg/kubelet/apis" kubeletapis "k8s.io/kubernetes/pkg/kubelet/apis"
vol "k8s.io/kubernetes/pkg/volume" vol "k8s.io/kubernetes/pkg/volume"
@ -53,11 +54,12 @@ var _ = admission.Interface(&persistentVolumeLabel{})
type persistentVolumeLabel struct { type persistentVolumeLabel struct {
*admission.Handler *admission.Handler
mutex sync.Mutex mutex sync.Mutex
ebsVolumes aws.Volumes cloudConfig []byte
cloudConfig []byte awsPVLabeler cloudprovider.PVLabeler
gceCloudProvider *gce.Cloud gcePVLabeler cloudprovider.PVLabeler
azureProvider *azure.Cloud azurePVLabeler cloudprovider.PVLabeler
openStackPVLabeler cloudprovider.PVLabeler
} }
var _ admission.MutationInterface = &persistentVolumeLabel{} var _ admission.MutationInterface = &persistentVolumeLabel{}
@ -186,47 +188,47 @@ func (l *persistentVolumeLabel) findAWSEBSLabels(volume *api.PersistentVolume) (
if volume.Spec.AWSElasticBlockStore.VolumeID == vol.ProvisionedVolumeName { if volume.Spec.AWSElasticBlockStore.VolumeID == vol.ProvisionedVolumeName {
return nil, nil return nil, nil
} }
ebsVolumes, err := l.getEBSVolumes() pvlabler, err := l.getAWSPVLabeler()
if err != nil { if err != nil {
return nil, err return nil, err
} }
if ebsVolumes == nil { if pvlabler == nil {
return nil, fmt.Errorf("unable to build AWS cloud provider for EBS") return nil, fmt.Errorf("unable to build AWS cloud provider for EBS")
} }
// TODO: GetVolumeLabels is actually a method on the Volumes interface pv := &v1.PersistentVolume{}
// If that gets standardized we can refactor to reduce code duplication err = k8s_api_v1.Convert_core_PersistentVolume_To_v1_PersistentVolume(volume, pv, nil)
spec := aws.KubernetesVolumeID(volume.Spec.AWSElasticBlockStore.VolumeID)
labels, err := ebsVolumes.GetVolumeLabels(spec)
if err != nil { if err != nil {
return nil, err return nil, fmt.Errorf("failed to convert PersistentVolume to core/v1: %q", err)
} }
return labels, nil return pvlabler.GetLabelsForVolume(context.TODO(), pv)
} }
// getEBSVolumes returns the AWS Volumes interface for ebs // getAWSPVLabeler returns the AWS implementation of PVLabeler
func (l *persistentVolumeLabel) getEBSVolumes() (aws.Volumes, error) { func (l *persistentVolumeLabel) getAWSPVLabeler() (cloudprovider.PVLabeler, error) {
l.mutex.Lock() l.mutex.Lock()
defer l.mutex.Unlock() defer l.mutex.Unlock()
if l.ebsVolumes == nil { if l.awsPVLabeler == nil {
var cloudConfigReader io.Reader var cloudConfigReader io.Reader
if len(l.cloudConfig) > 0 { if len(l.cloudConfig) > 0 {
cloudConfigReader = bytes.NewReader(l.cloudConfig) cloudConfigReader = bytes.NewReader(l.cloudConfig)
} }
cloudProvider, err := cloudprovider.GetCloudProvider("aws", cloudConfigReader) cloudProvider, err := cloudprovider.GetCloudProvider("aws", cloudConfigReader)
if err != nil || cloudProvider == nil { if err != nil || cloudProvider == nil {
return nil, err return nil, err
} }
awsCloudProvider, ok := cloudProvider.(*aws.Cloud)
awsPVLabeler, ok := cloudProvider.(cloudprovider.PVLabeler)
if !ok { if !ok {
// GetCloudProvider has gone very wrong return nil, errors.New("AWS cloud provider does not implement PV labeling")
return nil, fmt.Errorf("error retrieving AWS cloud provider")
} }
l.ebsVolumes = awsCloudProvider
l.awsPVLabeler = awsPVLabeler
} }
return l.ebsVolumes, nil return l.awsPVLabeler, nil
} }
func (l *persistentVolumeLabel) findGCEPDLabels(volume *api.PersistentVolume) (map[string]string, error) { func (l *persistentVolumeLabel) findGCEPDLabels(volume *api.PersistentVolume) (map[string]string, error) {
@ -235,72 +237,73 @@ func (l *persistentVolumeLabel) findGCEPDLabels(volume *api.PersistentVolume) (m
return nil, nil return nil, nil
} }
provider, err := l.getGCECloudProvider() pvlabler, err := l.getGCEPVLabeler()
if err != nil { if err != nil {
return nil, err return nil, err
} }
if provider == nil { if pvlabler == nil {
return nil, fmt.Errorf("unable to build GCE cloud provider for PD") return nil, fmt.Errorf("unable to build GCE cloud provider for PD")
} }
// If the zone is already labeled, honor the hint pv := &v1.PersistentVolume{}
zone := volume.Labels[kubeletapis.LabelZoneFailureDomain] err = k8s_api_v1.Convert_core_PersistentVolume_To_v1_PersistentVolume(volume, pv, nil)
labels, err := provider.GetAutoLabelsForPD(volume.Spec.GCEPersistentDisk.PDName, zone)
if err != nil { if err != nil {
return nil, err return nil, fmt.Errorf("failed to convert PersistentVolume to core/v1: %q", err)
} }
return pvlabler.GetLabelsForVolume(context.TODO(), pv)
return labels, nil
} }
// getGCECloudProvider returns the GCE cloud provider, for use for querying volume labels // getGCEPVLabeler returns the GCE implementation of PVLabeler
func (l *persistentVolumeLabel) getGCECloudProvider() (*gce.Cloud, error) { func (l *persistentVolumeLabel) getGCEPVLabeler() (cloudprovider.PVLabeler, error) {
l.mutex.Lock() l.mutex.Lock()
defer l.mutex.Unlock() defer l.mutex.Unlock()
if l.gceCloudProvider == nil { if l.gcePVLabeler == nil {
var cloudConfigReader io.Reader var cloudConfigReader io.Reader
if len(l.cloudConfig) > 0 { if len(l.cloudConfig) > 0 {
cloudConfigReader = bytes.NewReader(l.cloudConfig) cloudConfigReader = bytes.NewReader(l.cloudConfig)
} }
cloudProvider, err := cloudprovider.GetCloudProvider("gce", cloudConfigReader) cloudProvider, err := cloudprovider.GetCloudProvider("gce", cloudConfigReader)
if err != nil || cloudProvider == nil { if err != nil || cloudProvider == nil {
return nil, err return nil, err
} }
gceCloudProvider, ok := cloudProvider.(*gce.Cloud)
gcePVLabeler, ok := cloudProvider.(cloudprovider.PVLabeler)
if !ok { if !ok {
// GetCloudProvider has gone very wrong return nil, errors.New("GCE cloud provider does not implement PV labeling")
return nil, fmt.Errorf("error retrieving GCE cloud provider")
} }
l.gceCloudProvider = gceCloudProvider
l.gcePVLabeler = gcePVLabeler
} }
return l.gceCloudProvider, nil return l.gcePVLabeler, nil
} }
// getAzureCloudProvider returns the Azure cloud provider, for use for querying volume labels // getAzurePVLabeler returns the Azure implementation of PVLabeler
func (l *persistentVolumeLabel) getAzureCloudProvider() (*azure.Cloud, error) { func (l *persistentVolumeLabel) getAzurePVLabeler() (cloudprovider.PVLabeler, error) {
l.mutex.Lock() l.mutex.Lock()
defer l.mutex.Unlock() defer l.mutex.Unlock()
if l.azureProvider == nil { if l.azurePVLabeler == nil {
var cloudConfigReader io.Reader var cloudConfigReader io.Reader
if len(l.cloudConfig) > 0 { if len(l.cloudConfig) > 0 {
cloudConfigReader = bytes.NewReader(l.cloudConfig) cloudConfigReader = bytes.NewReader(l.cloudConfig)
} }
cloudProvider, err := cloudprovider.GetCloudProvider("azure", cloudConfigReader) cloudProvider, err := cloudprovider.GetCloudProvider("azure", cloudConfigReader)
if err != nil || cloudProvider == nil { if err != nil || cloudProvider == nil {
return nil, err return nil, err
} }
azureProvider, ok := cloudProvider.(*azure.Cloud)
azurePVLabeler, ok := cloudProvider.(cloudprovider.PVLabeler)
if !ok { if !ok {
// GetCloudProvider has gone very wrong return nil, errors.New("Azure cloud provider does not implement PV labeling")
return nil, fmt.Errorf("error retrieving Azure cloud provider")
} }
l.azureProvider = azureProvider l.azurePVLabeler = azurePVLabeler
} }
return l.azureProvider, nil return l.azurePVLabeler, nil
} }
func (l *persistentVolumeLabel) findAzureDiskLabels(volume *api.PersistentVolume) (map[string]string, error) { func (l *persistentVolumeLabel) findAzureDiskLabels(volume *api.PersistentVolume) (map[string]string, error) {
@ -309,13 +312,18 @@ func (l *persistentVolumeLabel) findAzureDiskLabels(volume *api.PersistentVolume
return nil, nil return nil, nil
} }
provider, err := l.getAzureCloudProvider() pvlabler, err := l.getAzurePVLabeler()
if err != nil { if err != nil {
return nil, err return nil, err
} }
if provider == nil { if pvlabler == nil {
return nil, fmt.Errorf("unable to build Azure cloud provider for AzureDisk") return nil, fmt.Errorf("unable to build Azure cloud provider for AzureDisk")
} }
return provider.GetAzureDiskLabels(volume.Spec.AzureDisk.DataDiskURI) pv := &v1.PersistentVolume{}
err = k8s_api_v1.Convert_core_PersistentVolume_To_v1_PersistentVolume(volume, pv, nil)
if err != nil {
return nil, fmt.Errorf("failed to convert PersistentVolume to core/v1: %q", err)
}
return pvlabler.GetLabelsForVolume(context.TODO(), pv)
} }

60
plugin/pkg/admission/storage/persistentvolume/label/admission_test.go
View File

@ -17,18 +17,17 @@ limitations under the License.
package label package label
import ( import (
"testing" "context"
"fmt" "fmt"
"reflect" "reflect"
"sort" "sort"
"testing"
"k8s.io/apimachinery/pkg/api/resource" "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apiserver/pkg/admission" "k8s.io/apiserver/pkg/admission"
cloudprovider "k8s.io/cloud-provider"
api "k8s.io/kubernetes/pkg/apis/core" api "k8s.io/kubernetes/pkg/apis/core"
"k8s.io/kubernetes/pkg/cloudprovider/providers/aws"
kubeletapis "k8s.io/kubernetes/pkg/kubelet/apis" kubeletapis "k8s.io/kubernetes/pkg/kubelet/apis"
volumeutil "k8s.io/kubernetes/pkg/volume/util" volumeutil "k8s.io/kubernetes/pkg/volume/util"
) )
@ -38,47 +37,12 @@ type mockVolumes struct {
volumeLabelsError error volumeLabelsError error
} }
var _ aws.Volumes = &mockVolumes{} var _ cloudprovider.PVLabeler = &mockVolumes{}
func (v *mockVolumes) AttachDisk(diskName aws.KubernetesVolumeID, nodeName types.NodeName) (string, error) { func (v *mockVolumes) GetLabelsForVolume(ctx context.Context, pv *v1.PersistentVolume) (map[string]string, error) {
return "", fmt.Errorf("not implemented")
}
func (v *mockVolumes) DetachDisk(diskName aws.KubernetesVolumeID, nodeName types.NodeName) (string, error) {
return "", fmt.Errorf("not implemented")
}
func (v *mockVolumes) CreateDisk(volumeOptions *aws.VolumeOptions) (volumeName aws.KubernetesVolumeID, err error) {
return "", fmt.Errorf("not implemented")
}
func (v *mockVolumes) DeleteDisk(volumeName aws.KubernetesVolumeID) (bool, error) {
return false, fmt.Errorf("not implemented")
}
func (v *mockVolumes) GetVolumeLabels(volumeName aws.KubernetesVolumeID) (map[string]string, error) {
return v.volumeLabels, v.volumeLabelsError return v.volumeLabels, v.volumeLabelsError
} }
func (v *mockVolumes) GetDiskPath(volumeName aws.KubernetesVolumeID) (string, error) {
return "", fmt.Errorf("not implemented")
}
func (v *mockVolumes) DiskIsAttached(volumeName aws.KubernetesVolumeID, nodeName types.NodeName) (bool, error) {
return false, fmt.Errorf("not implemented")
}
func (v *mockVolumes) DisksAreAttached(nodeDisks map[types.NodeName][]aws.KubernetesVolumeID) (map[types.NodeName]map[aws.KubernetesVolumeID]bool, error) {
return nil, fmt.Errorf("not implemented")
}
func (v *mockVolumes) ResizeDisk(
diskName aws.KubernetesVolumeID,
oldSize resource.Quantity,
newSize resource.Quantity) (resource.Quantity, error) {
return oldSize, nil
}
func mockVolumeFailure(err error) *mockVolumes { func mockVolumeFailure(err error) *mockVolumes {
return &mockVolumes{volumeLabelsError: err} return &mockVolumes{volumeLabelsError: err}
} }
@ -135,7 +99,7 @@ func TestAdmission(t *testing.T) {
} }
// Errors from the cloudprovider block creation of the volume // Errors from the cloudprovider block creation of the volume
pvHandler.ebsVolumes = mockVolumeFailure(fmt.Errorf("invalid volume")) pvHandler.awsPVLabeler = mockVolumeFailure(fmt.Errorf("invalid volume"))
err = handler.Admit(admission.NewAttributesRecord(&awsPV, nil, api.Kind("PersistentVolume").WithVersion("version"), awsPV.Namespace, awsPV.Name, api.Resource("persistentvolumes").WithVersion("version"), "", admission.Create, false, nil)) err = handler.Admit(admission.NewAttributesRecord(&awsPV, nil, api.Kind("PersistentVolume").WithVersion("version"), awsPV.Namespace, awsPV.Name, api.Resource("persistentvolumes").WithVersion("version"), "", admission.Create, false, nil))
if err == nil { if err == nil {
t.Errorf("Expected error when aws pv info fails") t.Errorf("Expected error when aws pv info fails")
@ -143,7 +107,7 @@ func TestAdmission(t *testing.T) {
// Don't add labels if the cloudprovider doesn't return any // Don't add labels if the cloudprovider doesn't return any
labels := make(map[string]string) labels := make(map[string]string)
pvHandler.ebsVolumes = mockVolumeLabels(labels) pvHandler.awsPVLabeler = mockVolumeLabels(labels)
err = handler.Admit(admission.NewAttributesRecord(&awsPV, nil, api.Kind("PersistentVolume").WithVersion("version"), awsPV.Namespace, awsPV.Name, api.Resource("persistentvolumes").WithVersion("version"), "", admission.Create, false, nil)) err = handler.Admit(admission.NewAttributesRecord(&awsPV, nil, api.Kind("PersistentVolume").WithVersion("version"), awsPV.Namespace, awsPV.Name, api.Resource("persistentvolumes").WithVersion("version"), "", admission.Create, false, nil))
if err != nil { if err != nil {
t.Errorf("Expected no error when creating aws pv") t.Errorf("Expected no error when creating aws pv")
@ -156,7 +120,7 @@ func TestAdmission(t *testing.T) {
} }
// Don't panic if the cloudprovider returns nil, nil // Don't panic if the cloudprovider returns nil, nil
pvHandler.ebsVolumes = mockVolumeFailure(nil) pvHandler.awsPVLabeler = mockVolumeFailure(nil)
err = handler.Admit(admission.NewAttributesRecord(&awsPV, nil, api.Kind("PersistentVolume").WithVersion("version"), awsPV.Namespace, awsPV.Name, api.Resource("persistentvolumes").WithVersion("version"), "", admission.Create, false, nil)) err = handler.Admit(admission.NewAttributesRecord(&awsPV, nil, api.Kind("PersistentVolume").WithVersion("version"), awsPV.Namespace, awsPV.Name, api.Resource("persistentvolumes").WithVersion("version"), "", admission.Create, false, nil))
if err != nil { if err != nil {
t.Errorf("Expected no error when cloud provider returns empty labels") t.Errorf("Expected no error when cloud provider returns empty labels")
@ -168,7 +132,7 @@ func TestAdmission(t *testing.T) {
labels["b"] = "2" labels["b"] = "2"
zones, _ := volumeutil.ZonesToSet("1,2,3") zones, _ := volumeutil.ZonesToSet("1,2,3")
labels[kubeletapis.LabelZoneFailureDomain] = volumeutil.ZonesSetToLabelValue(zones) labels[kubeletapis.LabelZoneFailureDomain] = volumeutil.ZonesSetToLabelValue(zones)
pvHandler.ebsVolumes = mockVolumeLabels(labels) pvHandler.awsPVLabeler = mockVolumeLabels(labels)
err = handler.Admit(admission.NewAttributesRecord(&awsPV, nil, api.Kind("PersistentVolume").WithVersion("version"), awsPV.Namespace, awsPV.Name, api.Resource("persistentvolumes").WithVersion("version"), "", admission.Create, false, nil)) err = handler.Admit(admission.NewAttributesRecord(&awsPV, nil, api.Kind("PersistentVolume").WithVersion("version"), awsPV.Namespace, awsPV.Name, api.Resource("persistentvolumes").WithVersion("version"), "", admission.Create, false, nil))
if err != nil { if err != nil {
t.Errorf("Expected no error when creating aws pv") t.Errorf("Expected no error when creating aws pv")
@ -223,7 +187,7 @@ func TestAdmission(t *testing.T) {
labels["a"] = "1" labels["a"] = "1"
labels["b"] = "2" labels["b"] = "2"
labels["c"] = "3" labels["c"] = "3"
pvHandler.ebsVolumes = mockVolumeLabels(labels) pvHandler.awsPVLabeler = mockVolumeLabels(labels)
err = handler.Admit(admission.NewAttributesRecord(&awsPV, nil, api.Kind("PersistentVolume").WithVersion("version"), awsPV.Namespace, awsPV.Name, api.Resource("persistentvolumes").WithVersion("version"), "", admission.Create, false, nil)) err = handler.Admit(admission.NewAttributesRecord(&awsPV, nil, api.Kind("PersistentVolume").WithVersion("version"), awsPV.Namespace, awsPV.Name, api.Resource("persistentvolumes").WithVersion("version"), "", admission.Create, false, nil))
if err != nil { if err != nil {
t.Errorf("Expected no error when creating aws pv") t.Errorf("Expected no error when creating aws pv")
@ -244,7 +208,7 @@ func TestAdmission(t *testing.T) {
labels["e"] = "1" labels["e"] = "1"
labels["f"] = "2" labels["f"] = "2"
labels["g"] = "3" labels["g"] = "3"
pvHandler.ebsVolumes = mockVolumeLabels(labels) pvHandler.awsPVLabeler = mockVolumeLabels(labels)
err = handler.Admit(admission.NewAttributesRecord(&awsPV, nil, api.Kind("PersistentVolume").WithVersion("version"), awsPV.Namespace, awsPV.Name, api.Resource("persistentvolumes").WithVersion("version"), "", admission.Create, false, nil)) err = handler.Admit(admission.NewAttributesRecord(&awsPV, nil, api.Kind("PersistentVolume").WithVersion("version"), awsPV.Namespace, awsPV.Name, api.Resource("persistentvolumes").WithVersion("version"), "", admission.Create, false, nil))
if err != nil { if err != nil {
t.Errorf("Expected no error when creating aws pv") t.Errorf("Expected no error when creating aws pv")