Provision vsphere volumes as per zone and attach labels

pkg/cloudprovider/providers/vsphere/nodemanager.go

@@ -22,10 +22,13 @@ import (
 	"strings"
 	"sync"
 
+	"github.com/vmware/govmomi/object"
 	"k8s.io/api/core/v1"
 	k8stypes "k8s.io/apimachinery/pkg/types"
+	cloudprovider "k8s.io/cloud-provider"
 	"k8s.io/klog"
 	"k8s.io/kubernetes/pkg/cloudprovider/providers/vsphere/vclib"
+	kubeletapis "k8s.io/kubernetes/pkg/kubelet/apis"
 )
 
 // Stores info about the kubernetes node
@@ -47,17 +50,25 @@ type NodeManager struct {
 	registeredNodes map[string]*v1.Node
 	//CredentialsManager
 	credentialManager *SecretCredentialManager
+	// Maps node name to zone information
+	zoneInfoMap map[string]*cloudprovider.Zone
+	// zone tag category from vsphere.conf
+	zoneCategoryName string
+	// region tag category from vsphere.conf
+	regionCategoryName string
 
 	// Mutexes
 	registeredNodesLock   sync.RWMutex
 	nodeInfoLock          sync.RWMutex
 	credentialManagerLock sync.Mutex
+	zoneInfoLock          sync.RWMutex
 }
 
 type NodeDetails struct {
 	NodeName string
 	vm       *vclib.VirtualMachine
 	VMUUID   string
+	Zone     cloudprovider.Zone
 }
 
 // TODO: Make it configurable in vsphere.conf
@@ -190,6 +201,12 @@ func (nm *NodeManager) DiscoverNode(node *v1.Node) error {
 					klog.V(4).Infof("Found node %s as vm=%+v in vc=%s and datacenter=%s",
 						node.Name, vm, res.vc, res.datacenter.Name())
 
+					// keep the zones discovered for the node
+					nodeZone := node.ObjectMeta.Labels[kubeletapis.LabelZoneFailureDomain]
+					nodeRegion := node.ObjectMeta.Labels[kubeletapis.LabelZoneRegion]
+					zone := cloudprovider.Zone{FailureDomain: nodeZone, Region: nodeRegion}
+					klog.V(4).Infof("Storing zone %v for node %s", zone, node.Name)
+					nm.addZoneInfo(ctx, node.ObjectMeta.Name, &zone)
 					nodeInfo := &NodeInfo{dataCenter: res.datacenter, vm: vm, vcServer: res.vc, vmUUID: nodeUUID}
 					nm.addNodeInfo(node.ObjectMeta.Name, nodeInfo)
 					for range queueChannel {
@@ -214,6 +231,20 @@ func (nm *NodeManager) DiscoverNode(node *v1.Node) error {
 	return vclib.ErrNoVMFound
 }
 
+func (nm *NodeManager) addZoneInfo(ctx context.Context, nodeName string, zone *cloudprovider.Zone) {
+	klog.V(4).Infof("add zone info to node %s: %v", nodeName, *zone)
+	nm.zoneInfoLock.Lock()
+	nm.zoneInfoMap[nodeName] = zone
+	nm.zoneInfoLock.Unlock()
+}
+
+func (nm *NodeManager) getZoneInfo(ctx context.Context, nodeName string) (*cloudprovider.Zone, error) {
+	nm.zoneInfoLock.RLock()
+	zone := nm.zoneInfoMap[nodeName]
+	nm.zoneInfoLock.RUnlock()
+	return zone, nil
+}
+
 func (nm *NodeManager) RegisterNode(node *v1.Node) error {
 	nm.addNode(node)
 	nm.DiscoverNode(node)
@@ -258,6 +289,10 @@ func (nm *NodeManager) removeNode(node *v1.Node) {
 	nm.nodeInfoLock.Lock()
 	delete(nm.nodeInfoMap, node.ObjectMeta.Name)
 	nm.nodeInfoLock.Unlock()
+
+	nm.zoneInfoLock.Lock()
+	delete(nm.zoneInfoMap, node.ObjectMeta.Name)
+	nm.zoneInfoLock.Unlock()
 }
 
 // GetNodeInfo returns a NodeInfo which datacenter, vm and vc server ip address.
@@ -310,7 +345,11 @@ func (nm *NodeManager) GetNodeDetails() ([]NodeDetails, error) {
 			return nil, err
 		}
 		klog.V(4).Infof("Updated NodeInfo %v for node %q.", nodeInfo, nodeName)
-		nodeDetails = append(nodeDetails, NodeDetails{nodeName, nodeInfo.vm, nodeInfo.vmUUID})
+		zone, err := nm.getZoneInfo(context.TODO(), nodeName)
+		if err != nil {
+			return nil, err
+		}
+		nodeDetails = append(nodeDetails, NodeDetails{nodeName, nodeInfo.vm, nodeInfo.vmUUID, *zone})
 	}
 	return nodeDetails, nil
 }
@@ -443,3 +482,27 @@ func (nm *NodeManager) UpdateCredentialManager(credentialManager *SecretCredenti
 	defer nm.credentialManagerLock.Unlock()
 	nm.credentialManager = credentialManager
 }
+
+func (nm *NodeManager) GetHostsInZone(ctx context.Context, zone cloudprovider.Zone) ([]*object.HostSystem, error) {
+	klog.V(7).Infof("GetHostsInZone called with registeredNodes: %v", nm.registeredNodes)
+	nodeDetails, err := nm.GetNodeDetails()
+	if err != nil {
+		return nil, err
+	}
+	klog.V(4).Infof("Node Details: %v", nodeDetails)
+	// Return those hosts that are in given zone.
+	hosts := make([]*object.HostSystem, 0)
+	for _, n := range nodeDetails {
+		// Match the provided zone with the node's zone, if not empty.
+		if (zone.FailureDomain == "" || zone.FailureDomain == n.Zone.FailureDomain) && (zone.Region == "" || zone.Region == n.Zone.Region) {
+			host, err := n.vm.HostSystem(ctx)
+			if err != nil {
+				klog.Errorf("Failed to get host system for VM %s. err: %+v", n.vm, err)
+				continue
+			}
+			hosts = append(hosts, host)
+		}
+	}
+	klog.V(4).Infof("GetHostsInZone %v returning: %v", zone, hosts)
+	return hosts, nil
+}

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

@@ -145,6 +145,20 @@ func (dc *Datacenter) GetAllDatastores(ctx context.Context) (map[string]*Datasto
 	return dsURLInfoMap, nil
 }
 
+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)
 }
+
+// Get 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.Error("Failed to retrieve datastore host mount property. err: %v", err)
+		return nil, err
+	}
+	hosts := make([]types.ManagedObjectReference, 0)
+	for _, dsHostMount := range dsMo.Host {
+		hosts = append(hosts, dsHostMount.Key)
+	}
+	return hosts, nil
+}

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

@@ -33,6 +33,7 @@ type VolumeOptions struct {
 	StoragePolicyName      string
 	StoragePolicyID        string
 	SCSIControllerType     string
+	Zone                   []string
 }
 
 var (