Applies zone labels to newly created vsphere volumes

pkg/cloudprovider/providers/vsphere/BUILD

@@ -21,12 +21,15 @@ go_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",
         "//staging/src/k8s.io/cloud-provider/node/helpers:go_default_library",
+        "//staging/src/k8s.io/cloud-provider/volume/helpers: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",
@@ -58,6 +61,7 @@ go_test(
         "//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: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",
@@ -67,6 +71,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

@@ -40,15 +40,18 @@ const (
 
 // Other Constants
 const (
-	LogLevel                 = 4
-	DatastoreProperty        = "datastore"
-	ResourcePoolProperty     = "resourcePool"
-	DatastoreInfoProperty    = "info"
-	VirtualMachineType       = "VirtualMachine"
-	RoundTripperDefaultCount = 3
-	VSANDatastoreType        = "vsan"
-	DummyVMPrefixName        = "vsphere-k8s"
-	ActivePowerState         = "poweredOn"
+	LogLevel                   = 4
+	DatastoreProperty          = "datastore"
+	ResourcePoolProperty       = "resourcePool"
+	DatastoreInfoProperty      = "info"
+	VirtualMachineType         = "VirtualMachine"
+	RoundTripperDefaultCount   = 3
+	VSANDatastoreType          = "vsan"
+	DummyVMPrefixName          = "vsphere-k8s"
+	ActivePowerState           = "poweredOn"
+	DatacenterType             = "Datacenter"
+	ClusterComputeResourceType = "ClusterComputeResource"
+	HostSystemType             = "HostSystem"
 )
 
 // Test Constants

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

@@ -148,6 +148,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/vsphere_volume.go

@@ -27,6 +27,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	volumehelpers "k8s.io/cloud-provider/volume/helpers"
 	"k8s.io/klog"
 	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/util/mount"
@@ -411,6 +412,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 == v1.LabelZoneFailureDomain {
+				values, err = volumehelpers.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

@@ -72,6 +72,7 @@ type VolumeSpec struct {
 	Fstype            string
 	StoragePolicyID   string
 	StoragePolicyName string
+	Labels            map[string]string
 }
 
 func verifyDevicePath(path string) (string, error) {
@@ -148,12 +149,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/admission.go

@@ -59,6 +59,7 @@ type persistentVolumeLabel struct {
 	gcePVLabeler       cloudprovider.PVLabeler
 	azurePVLabeler     cloudprovider.PVLabeler
 	openStackPVLabeler cloudprovider.PVLabeler
+	vspherePVLabeler   cloudprovider.PVLabeler
 }
 
 var _ admission.MutationInterface = &persistentVolumeLabel{}
@@ -138,7 +139,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 {
 		if volume.Labels == nil {
@@ -384,3 +391,48 @@ func (l *persistentVolumeLabel) findCinderDiskLabels(volume *api.PersistentVolum
 	return pvlabler.GetLabelsForVolume(context.TODO(), pv)
 
 }
+
+func (l *persistentVolumeLabel) findVsphereVolumeLabels(volume *api.PersistentVolume) (map[string]string, error) {
+	pvlabler, err := l.getVspherePVLabeler()
+	if err != nil {
+		return nil, err
+	}
+	if pvlabler == nil {
+		return nil, fmt.Errorf("unable to build vSphere cloud provider")
+	}
+
+	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)
+	}
+	labels, err := pvlabler.GetLabelsForVolume(context.TODO(), pv)
+	if err != nil {
+		return nil, err
+	}
+
+	return labels, nil
+}
+
+func (l *persistentVolumeLabel) getVspherePVLabeler() (cloudprovider.PVLabeler, error) {
+	l.mutex.Lock()
+	defer l.mutex.Unlock()
+
+	if l.vspherePVLabeler == 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
+		}
+		vspherePVLabeler, ok := cloudProvider.(cloudprovider.PVLabeler)
+		if !ok {
+			// GetCloudProvider failed
+			return nil, errors.New("vSphere Cloud Provider does not implement PV labeling")
+		}
+		l.vspherePVLabeler = vspherePVLabeler
+	}
+	return l.vspherePVLabeler, nil
+}

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

@@ -683,6 +683,72 @@ func Test_PVLAdmission(t *testing.T) {
 			},
 			err: nil,
 		},
+		{
+			name:    "vSphere PV labeled correctly",
+			handler: newPersistentVolumeLabel(),
+			pvlabeler: mockVolumeLabels(map[string]string{
+				"a":                       "1",
+				"b":                       "2",
+				v1.LabelZoneFailureDomain: "1__2__3",
+			}),
+			preAdmissionPV: &api.PersistentVolume{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "vSpherePV",
+					Namespace: "myns",
+				},
+				Spec: api.PersistentVolumeSpec{
+					PersistentVolumeSource: api.PersistentVolumeSource{
+						VsphereVolume: &api.VsphereVirtualDiskVolumeSource{
+							VolumePath: "123",
+						},
+					},
+				},
+			},
+			postAdmissionPV: &api.PersistentVolume{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "vSpherePV",
+					Namespace: "myns",
+					Labels: map[string]string{
+						"a":                       "1",
+						"b":                       "2",
+						v1.LabelZoneFailureDomain: "1__2__3",
+					},
+				},
+				Spec: api.PersistentVolumeSpec{
+					PersistentVolumeSource: api.PersistentVolumeSource{
+						VsphereVolume: &api.VsphereVirtualDiskVolumeSource{
+							VolumePath: "123",
+						},
+					},
+					NodeAffinity: &api.VolumeNodeAffinity{
+						Required: &api.NodeSelector{
+							NodeSelectorTerms: []api.NodeSelectorTerm{
+								{
+									MatchExpressions: []api.NodeSelectorRequirement{
+										{
+											Key:      "a",
+											Operator: api.NodeSelectorOpIn,
+											Values:   []string{"1"},
+										},
+										{
+											Key:      "b",
+											Operator: api.NodeSelectorOpIn,
+											Values:   []string{"2"},
+										},
+										{
+											Key:      v1.LabelZoneFailureDomain,
+											Operator: api.NodeSelectorOpIn,
+											Values:   []string{"1", "2", "3"},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			err: nil,
+		},
 	}
 
 	for _, testcase := range testcases {
@@ -718,6 +784,7 @@ func setPVLabeler(handler *persistentVolumeLabel, pvlabeler cloudprovider.PVLabe
 	handler.gcePVLabeler = pvlabeler
 	handler.azurePVLabeler = pvlabeler
 	handler.openStackPVLabeler = pvlabeler
+	handler.vspherePVLabeler = pvlabeler
 }
 
 // sortMatchExpressions sorts a PV's node selector match expressions by key name if it is not nil