Applies zone labels to newly created vsphere volumes

pkg/cloudprovider/providers/vsphere/BUILD

@@ -19,18 +19,22 @@ go_library(
         "//pkg/apis/core/v1/helper:go_default_library",
         "//pkg/cloudprovider/providers/vsphere/vclib:go_default_library",
         "//pkg/cloudprovider/providers/vsphere/vclib/diskmanagers:go_default_library",
+        "//pkg/volume/util:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/version:go_default_library",
         "//staging/src/k8s.io/client-go/informers:go_default_library",
         "//staging/src/k8s.io/client-go/listers/core/v1:go_default_library",
         "//staging/src/k8s.io/client-go/tools/cache:go_default_library",
         "//staging/src/k8s.io/cloud-provider:go_default_library",
+        "//vendor/github.com/vmware/govmomi/object:go_default_library",
         "//vendor/github.com/vmware/govmomi/vapi/rest:go_default_library",
         "//vendor/github.com/vmware/govmomi/vapi/tags:go_default_library",
         "//vendor/github.com/vmware/govmomi/vim25:go_default_library",
         "//vendor/github.com/vmware/govmomi/vim25/mo:go_default_library",
+        "//vendor/github.com/vmware/govmomi/vim25/types:go_default_library",
         "//vendor/gopkg.in/gcfg.v1:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",
     ],
@@ -55,6 +59,7 @@ go_test(
         "//staging/src/k8s.io/client-go/informers:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes/fake:go_default_library",
         "//staging/src/k8s.io/cloud-provider:go_default_library",
+        "//vendor/github.com/vmware/govmomi/find:go_default_library",
         "//vendor/github.com/vmware/govmomi/lookup/simulator:go_default_library",
         "//vendor/github.com/vmware/govmomi/property:go_default_library",
         "//vendor/github.com/vmware/govmomi/simulator:go_default_library",
@@ -64,6 +69,7 @@ go_test(
         "//vendor/github.com/vmware/govmomi/vapi/simulator:go_default_library",
         "//vendor/github.com/vmware/govmomi/vapi/tags:go_default_library",
         "//vendor/github.com/vmware/govmomi/vim25/mo:go_default_library",
+        "//vendor/github.com/vmware/govmomi/vim25/types:go_default_library",
     ],
 )
 

pkg/cloudprovider/providers/vsphere/vclib/constants.go

@@ -49,6 +49,9 @@ const (
 	VSANDatastoreType          = "vsan"
 	DummyVMPrefixName          = "vsphere-k8s"
 	ActivePowerState           = "poweredOn"
+	DatacenterType             = "Datacenter"
+	ClusterComputeResourceType = "ClusterComputeResource"
+	HostSystemType             = "HostSystem"
 )
 
 // Test Constants

pkg/cloudprovider/providers/vsphere/vclib/datacenter.go

@@ -145,6 +145,21 @@ func (dc *Datacenter) GetAllDatastores(ctx context.Context) (map[string]*Datasto
 	return dsURLInfoMap, nil
 }
 
+// GetAllHosts returns all the host objects in this datacenter of VC
+func (dc *Datacenter) GetAllHosts(ctx context.Context) ([]types.ManagedObjectReference, error) {
+	finder := getFinder(dc)
+	hostSystems, err := finder.HostSystemList(ctx, "*")
+	if err != nil {
+		klog.Errorf("Failed to get all hostSystems. err: %+v", err)
+		return nil, err
+	}
+	var hostMors []types.ManagedObjectReference
+	for _, hs := range hostSystems {
+		hostMors = append(hostMors, hs.Reference())
+	}
+	return hostMors, nil
+}
+
 // GetDatastoreByPath gets the Datastore object from the given vmDiskPath
 func (dc *Datacenter) GetDatastoreByPath(ctx context.Context, vmDiskPath string) (*Datastore, error) {
 	datastorePathObj := new(object.DatastorePath)

pkg/cloudprovider/providers/vsphere/vclib/datastore.go

@@ -85,3 +85,19 @@ func (ds *Datastore) IsCompatibleWithStoragePolicy(ctx context.Context, storageP
 	}
 	return pbmClient.IsDatastoreCompatible(ctx, storagePolicyID, ds)
 }
+
+// GetDatastoreHostMounts gets the host names mounted on given datastore
+func (ds *Datastore) GetDatastoreHostMounts(ctx context.Context) ([]types.ManagedObjectReference, error) {
+	var dsMo mo.Datastore
+	pc := property.DefaultCollector(ds.Client())
+	err := pc.RetrieveOne(ctx, ds.Datastore.Reference(), []string{"host"}, &dsMo)
+	if err != nil {
+		klog.Errorf("Failed to retrieve datastore host mount property. err: %v", err)
+		return nil, err
+	}
+	hosts := make([]types.ManagedObjectReference, len(dsMo.Host))
+	for _, dsHostMount := range dsMo.Host {
+		hosts = append(hosts, dsHostMount.Key)
+	}
+	return hosts, nil
+}

pkg/cloudprovider/providers/vsphere/vsphere_test.go

@@ -23,10 +23,13 @@ import (
 	"io/ioutil"
 	"log"
 	"os"
+	"reflect"
+	"sort"
 	"strconv"
 	"strings"
 	"testing"
 
+	"github.com/vmware/govmomi/find"
 	lookup "github.com/vmware/govmomi/lookup/simulator"
 	"github.com/vmware/govmomi/property"
 	"github.com/vmware/govmomi/simulator"
@@ -36,6 +39,7 @@ import (
 	vapi "github.com/vmware/govmomi/vapi/simulator"
 	"github.com/vmware/govmomi/vapi/tags"
 	"github.com/vmware/govmomi/vim25/mo"
+	vmwaretypes "github.com/vmware/govmomi/vim25/types"
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/rand"
@@ -487,6 +491,192 @@ func TestZones(t *testing.T) {
 	}
 }
 
+func TestGetZoneToHosts(t *testing.T) {
+	// Common setup for all testcases in this test
+	ctx := context.TODO()
+
+	// Create a vcsim instance
+	cfg, cleanup := configFromSim()
+	defer cleanup()
+
+	// Create vSphere configuration object
+	vs, err := newControllerNode(cfg)
+	if err != nil {
+		t.Fatalf("Failed to construct/authenticate vSphere: %s", err)
+	}
+
+	// Configure region and zone categories
+	vs.cfg.Labels.Region = "k8s-region"
+	vs.cfg.Labels.Zone = "k8s-zone"
+
+	// Create vSphere client
+	vsi, ok := vs.vsphereInstanceMap[cfg.Global.VCenterIP]
+	if !ok {
+		t.Fatalf("Couldn't get vSphere instance: %s", cfg.Global.VCenterIP)
+	}
+
+	err = vsi.conn.Connect(ctx)
+	if err != nil {
+		t.Errorf("Failed to connect to vSphere: %s", err)
+	}
+
+	// Lookup Datacenter for this test's Workspace
+	dc, err := vclib.GetDatacenter(ctx, vsi.conn, vs.cfg.Workspace.Datacenter)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Property Collector instance
+	pc := property.DefaultCollector(vsi.conn.Client)
+
+	// find all hosts in VC
+	finder := find.NewFinder(vsi.conn.Client, true)
+	finder.SetDatacenter(dc.Datacenter)
+	allVcHostsList, err := finder.HostSystemList(ctx, "*")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var allVcHosts []vmwaretypes.ManagedObjectReference
+	for _, h := range allVcHostsList {
+		allVcHosts = append(allVcHosts, h.Reference())
+	}
+
+	// choose a cluster to apply zone/region tags
+	cluster := simulator.Map.Any("ClusterComputeResource")
+	var c mo.ClusterComputeResource
+	if err := pc.RetrieveOne(ctx, cluster.Reference(), []string{"host"}, &c); err != nil {
+		t.Fatal(err)
+	}
+
+	// choose one of the host inside this cluster to apply zone/region tags
+	if c.Host == nil || len(c.Host) == 0 {
+		t.Fatalf("This test needs a host inside a cluster.")
+	}
+	clusterHosts := c.Host
+	sortHosts(clusterHosts)
+	// pick the first host in the cluster to apply tags
+	host := clusterHosts[0]
+	remainingHostsInCluster := clusterHosts[1:]
+
+	// Tag manager instance
+	m := tags.NewManager(rest.NewClient(vsi.conn.Client))
+
+	// Create a region category
+	regionCat, err := m.CreateCategory(ctx, &tags.Category{Name: vs.cfg.Labels.Region})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Create a region tag
+	regionName := "k8s-region-US"
+	regionTag, err := m.CreateTag(ctx, &tags.Tag{CategoryID: regionCat, Name: regionName})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Create a zone category
+	zoneCat, err := m.CreateCategory(ctx, &tags.Category{Name: vs.cfg.Labels.Zone})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Create a zone tag
+	zone1Name := "k8s-zone-US-CA1"
+	zone1Tag, err := m.CreateTag(ctx, &tags.Tag{CategoryID: zoneCat, Name: zone1Name})
+	if err != nil {
+		t.Fatal(err)
+	}
+	zone1 := cloudprovider.Zone{FailureDomain: zone1Name, Region: regionName}
+
+	// Create a second zone tag
+	zone2Name := "k8s-zone-US-CA2"
+	zone2Tag, err := m.CreateTag(ctx, &tags.Tag{CategoryID: zoneCat, Name: zone2Name})
+	if err != nil {
+		t.Fatal(err)
+	}
+	zone2 := cloudprovider.Zone{FailureDomain: zone2Name, Region: regionName}
+
+	testcases := []struct {
+		name        string
+		tags        map[string][]mo.Reference
+		zoneToHosts map[cloudprovider.Zone][]vmwaretypes.ManagedObjectReference
+	}{
+		{
+			name:        "Zone and Region tags on host",
+			tags:        map[string][]mo.Reference{zone1Tag: {host}, regionTag: {host}},
+			zoneToHosts: map[cloudprovider.Zone][]vmwaretypes.ManagedObjectReference{zone1: {host}},
+		},
+		{
+			name:        "Zone on host Region on datacenter",
+			tags:        map[string][]mo.Reference{zone1Tag: {host}, regionTag: {dc}},
+			zoneToHosts: map[cloudprovider.Zone][]vmwaretypes.ManagedObjectReference{zone1: {host}},
+		},
+		{
+			name:        "Zone on cluster Region on datacenter",
+			tags:        map[string][]mo.Reference{zone1Tag: {cluster}, regionTag: {dc}},
+			zoneToHosts: map[cloudprovider.Zone][]vmwaretypes.ManagedObjectReference{zone1: clusterHosts},
+		},
+		{
+			name:        "Zone on cluster and override on host",
+			tags:        map[string][]mo.Reference{zone2Tag: {cluster}, zone1Tag: {host}, regionTag: {dc}},
+			zoneToHosts: map[cloudprovider.Zone][]vmwaretypes.ManagedObjectReference{zone1: {host}, zone2: remainingHostsInCluster},
+		},
+		{
+			name:        "Zone and Region on datacenter",
+			tags:        map[string][]mo.Reference{zone1Tag: {dc}, regionTag: {dc}},
+			zoneToHosts: map[cloudprovider.Zone][]vmwaretypes.ManagedObjectReference{zone1: allVcHosts},
+		},
+	}
+
+	for _, testcase := range testcases {
+		t.Run(testcase.name, func(t *testing.T) {
+			// apply tags to datacenter/cluster/host as per this testcase
+			for tagId, objects := range testcase.tags {
+				for _, object := range objects {
+					if err := m.AttachTag(ctx, tagId, object); err != nil {
+						t.Fatal(err)
+					}
+				}
+			}
+
+			// run the test
+			zoneToHosts, err := vs.GetZoneToHosts(ctx, vsi)
+			if err != nil {
+				t.Errorf("unexpected error when calling GetZoneToHosts: %q", err)
+			}
+
+			// do not depend on the sort order of hosts in result
+			sortHostsMap(zoneToHosts)
+			if !reflect.DeepEqual(zoneToHosts, testcase.zoneToHosts) {
+				t.Logf("expected result: %+v", testcase.zoneToHosts)
+				t.Logf("actual result: %+v", zoneToHosts)
+				t.Error("unexpected result from GetZoneToHosts")
+			}
+
+			// clean up tags applied on datacenter/cluster/host for this testcase
+			for tagId, objects := range testcase.tags {
+				for _, object := range objects {
+					if err = m.DetachTag(ctx, tagId, object); err != nil {
+						t.Fatal(err)
+					}
+				}
+			}
+		})
+	}
+}
+
+func sortHostsMap(zoneToHosts map[cloudprovider.Zone][]vmwaretypes.ManagedObjectReference) {
+	for _, hosts := range zoneToHosts {
+		sortHosts(hosts)
+	}
+}
+
+func sortHosts(hosts []vmwaretypes.ManagedObjectReference) {
+	sort.Slice(hosts, func(i, j int) bool {
+		return hosts[i].Value < hosts[j].Value
+	})
+}
+
 func TestInstances(t *testing.T) {
 	cfg, ok := configFromEnv()
 	if !ok {

pkg/volume/vsphere_volume/BUILD

@@ -31,6 +31,7 @@ go_library(
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//staging/src/k8s.io/cloud-provider:go_default_library",
+        "//vendor/k8s.io/cloud-provider/volume/helpers:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",
     ],
 )

pkg/volume/vsphere_volume/vsphere_volume.go

@@ -29,6 +29,7 @@ import (
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/klog"
 	"k8s.io/kubernetes/pkg/features"
+	"k8s.io/kubernetes/pkg/kubelet/apis"
 	"k8s.io/kubernetes/pkg/util/mount"
 	utilstrings "k8s.io/kubernetes/pkg/util/strings"
 	"k8s.io/kubernetes/pkg/volume"
@@ -407,6 +408,34 @@ func (v *vsphereVolumeProvisioner) Provision(selectedNode *v1.Node, allowedTopol
 		pv.Spec.AccessModes = v.plugin.GetAccessModes()
 	}
 
+	labels := volSpec.Labels
+	requirements := make([]v1.NodeSelectorRequirement, 0)
+	if len(labels) != 0 {
+		if pv.Labels == nil {
+			pv.Labels = make(map[string]string)
+		}
+		for k, v := range labels {
+			pv.Labels[k] = v
+			var values []string
+			if k == apis.LabelZoneFailureDomain {
+				values, err = util.LabelZonesToList(v)
+				if err != nil {
+					return nil, fmt.Errorf("failed to convert label string for Zone: %s to a List: %v", v, err)
+				}
+			} else {
+				values = []string{v}
+			}
+			requirements = append(requirements, v1.NodeSelectorRequirement{Key: k, Operator: v1.NodeSelectorOpIn, Values: values})
+		}
+	}
+
+	if len(requirements) > 0 {
+		pv.Spec.NodeAffinity = new(v1.VolumeNodeAffinity)
+		pv.Spec.NodeAffinity.Required = new(v1.NodeSelector)
+		pv.Spec.NodeAffinity.Required.NodeSelectorTerms = make([]v1.NodeSelectorTerm, 1)
+		pv.Spec.NodeAffinity.Required.NodeSelectorTerms[0].MatchExpressions = requirements
+	}
+
 	return pv, nil
 }
 

pkg/volume/vsphere_volume/vsphere_volume_util.go

@@ -70,6 +70,7 @@ type VolumeSpec struct {
 	Fstype            string
 	StoragePolicyID   string
 	StoragePolicyName string
+	Labels            map[string]string
 }
 
 func verifyDevicePath(path string) (string, error) {
@@ -147,12 +148,17 @@ func (util *VsphereDiskUtil) CreateVolume(v *vsphereVolumeProvisioner) (volSpec
 	if err != nil {
 		return nil, err
 	}
+	labels, err := cloud.GetVolumeLabels(vmDiskPath)
+	if err != nil {
+		return nil, err
+	}
 	volSpec = &VolumeSpec{
 		Path:              vmDiskPath,
 		Size:              volSizeKiB,
 		Fstype:            fstype,
 		StoragePolicyName: volumeOptions.StoragePolicyName,
 		StoragePolicyID:   volumeOptions.StoragePolicyID,
+		Labels:            labels,
 	}
 	klog.V(2).Infof("Successfully created vsphere volume %s", name)
 	return volSpec, nil

plugin/pkg/admission/storage/persistentvolume/label/BUILD

@@ -15,9 +15,11 @@ go_library(
     importpath = "k8s.io/kubernetes/plugin/pkg/admission/storage/persistentvolume/label",
     deps = [
         "//pkg/apis/core:go_default_library",
+        "//pkg/apis/core/v1:go_default_library",
         "//pkg/cloudprovider/providers/aws:go_default_library",
         "//pkg/cloudprovider/providers/azure:go_default_library",
         "//pkg/cloudprovider/providers/gce:go_default_library",
+        "//pkg/cloudprovider/providers/vsphere:go_default_library",
         "//pkg/features:go_default_library",
         "//pkg/kubeapiserver/admission:go_default_library",
         "//pkg/kubelet/apis:go_default_library",

plugin/pkg/admission/storage/persistentvolume/label/admission.go

@@ -30,6 +30,7 @@ import (
 	"k8s.io/kubernetes/pkg/cloudprovider/providers/aws"
 	"k8s.io/kubernetes/pkg/cloudprovider/providers/azure"
 	"k8s.io/kubernetes/pkg/cloudprovider/providers/gce"
+	"k8s.io/kubernetes/pkg/cloudprovider/providers/vsphere"
 	"k8s.io/kubernetes/pkg/features"
 	kubeapiserveradmission "k8s.io/kubernetes/pkg/kubeapiserver/admission"
 	kubeletapis "k8s.io/kubernetes/pkg/kubelet/apis"
@@ -60,6 +61,7 @@ type persistentVolumeLabel struct {
 	cloudConfig      []byte
 	gceCloudProvider *gce.Cloud
 	azureProvider    *azure.Cloud
+	vsphereProvider  *vsphere.VSphere
 }
 
 var _ admission.MutationInterface = &persistentVolumeLabel{}
@@ -132,6 +134,13 @@ func (l *persistentVolumeLabel) Admit(a admission.Attributes) (err error) {
 		}
 		volumeLabels = labels
 	}
+	if volume.Spec.VsphereVolume != nil {
+		labels, err := l.findVsphereVolumeLabels(volume)
+		if err != nil {
+			return admission.NewForbidden(a, fmt.Errorf("error querying vSphere Volume %s: %v", volume.Spec.VsphereVolume.VolumePath, err))
+		}
+		volumeLabels = labels
+	}
 
 	requirements := make([]api.NodeSelectorRequirement, 0)
 	if len(volumeLabels) != 0 {
@@ -323,3 +332,43 @@ func (l *persistentVolumeLabel) findAzureDiskLabels(volume *api.PersistentVolume
 
 	return provider.GetAzureDiskLabels(volume.Spec.AzureDisk.DataDiskURI)
 }
+
+func (l *persistentVolumeLabel) findVsphereVolumeLabels(volume *api.PersistentVolume) (map[string]string, error) {
+	pvlabler, err := l.getVSphereProvider()
+	if err != nil {
+		return nil, err
+	}
+	if pvlabler == nil {
+		return nil, fmt.Errorf("unable to build vSphere cloud provider")
+	}
+
+	labels, err := pvlabler.GetVolumeLabels(volume.Spec.VsphereVolume.VolumePath)
+	if err != nil {
+		return nil, err
+	}
+
+	return labels, nil
+}
+
+func (l *persistentVolumeLabel) getVSphereProvider() (*vsphere.VSphere, error) {
+	l.mutex.Lock()
+	defer l.mutex.Unlock()
+
+	if l.vsphereProvider == nil {
+		var cloudConfigReader io.Reader
+		if len(l.cloudConfig) > 0 {
+			cloudConfigReader = bytes.NewReader(l.cloudConfig)
+		}
+		cloudProvider, err := cloudprovider.GetCloudProvider("vsphere", cloudConfigReader)
+		if err != nil || cloudProvider == nil {
+			return nil, err
+		}
+		vsphereProvider, ok := cloudProvider.(*vsphere.VSphere)
+		if !ok {
+			// GetCloudProvider failed
+			return nil, fmt.Errorf("Error retrieving vSphere Cloud Provider")
+		}
+		l.vsphereProvider = vsphereProvider
+	}
+	return l.vsphereProvider, nil
+}