Support Zone-Aware Topologies in the KubeVirt CSI Driver

cmd/kubevirt-csi-driver/kubevirt-csi-driver.go

@@ -105,14 +105,20 @@ func handle() {
 		klog.Fatal(err)
 	}
 
-	var nodeID string
+	var (
+		nodeID            string
+		allowedTopologies = map[string]string{}
+	)
+
 	if *nodeName != "" {
 		node, err := tenantClientSet.CoreV1().Nodes().Get(context.TODO(), *nodeName, v1.GetOptions{})
 		if err != nil {
 			klog.Fatal(fmt.Errorf("failed to find node by name %v: %v", nodeName, err))
 		}
 		// systemUUID is the VM ID
 		nodeID = node.Status.NodeInfo.SystemUUID
+		allowedTopologies[service.WellKnownZoneTopologyKey] = node.Labels[service.WellKnownZoneTopologyKey]
+		allowedTopologies[service.WellKnownRegionTopologyKey] = node.Labels[service.WellKnownRegionTopologyKey]
 		klog.Infof("Node name: %v, Node ID: %s", nodeName, nodeID)
 	}
 
@@ -131,7 +137,8 @@ func handle() {
 		storageClassEnforcement,
 		nodeID,
 		*runNodeService,
-		*runControllerService)
+		*runControllerService,
+		allowedTopologies)
 
 	driver.Run(*endpoint)
 }

deploy/controller-infra/base/deploy.yaml

@@ -82,6 +82,7 @@ spec:
             - "--v=5"
             - "--timeout=3m"
             - "--retry-interval-max=1m"
+            - "--feature-gates=Topology=true"
           env:
             - name: ADDRESS
               value: /var/lib/csi/sockets/pluginproxy/csi.sock

deploy/controller-tenant/base/deploy.yaml

@@ -81,6 +81,7 @@ spec:
             - "--v=5"
             - "--timeout=3m"
             - "--retry-interval-max=1m"
+            - "--feature-gates=Topology=true"
           env:
             - name: ADDRESS
               value: /var/lib/csi/sockets/pluginproxy/csi.sock

pkg/service/controller.go

@@ -218,8 +218,12 @@ func (c *ControllerService) CreateVolume(ctx context.Context, req *csi.CreateVol
 	// Prepare serial for disk
 	serial := string(dv.GetUID())
 
-	// Return response
-	return &csi.CreateVolumeResponse{
+	zone, region, err := c.getAllowedZoneAndRegion(ctx, req.AccessibilityRequirements)
+	if err != nil {
+		return nil, err
+	}
+
+	cvr := &csi.CreateVolumeResponse{
 		Volume: &csi.Volume{
 			CapacityBytes: storageSize,
 			VolumeId:      dvName,
@@ -229,7 +233,27 @@ func (c *ControllerService) CreateVolume(ctx context.Context, req *csi.CreateVol
 			},
 			ContentSource: req.GetVolumeContentSource(),
 		},
-	}, nil
+	}
+
+	var allowedTopologies = map[string]string{}
+	if zone != "" {
+		allowedTopologies[WellKnownZoneTopologyKey] = zone
+	}
+
+	if region != "" {
+		allowedTopologies[WellKnownRegionTopologyKey] = region
+	}
+
+	if len(allowedTopologies) > 0 {
+		cvr.Volume.AccessibleTopology = []*csi.Topology{
+			{
+				Segments: allowedTopologies,
+			},
+		}
+	}
+
+	// Return response
+	return cvr, nil
 }
 
 func (c *ControllerService) determineDvSource(ctx context.Context, req *csi.CreateVolumeRequest) (*cdiv1.DataVolumeSource, error) {
@@ -768,3 +792,38 @@ func (c *ControllerService) getVMNameByCSINodeID(ctx context.Context, nodeID str
 
 	return "", status.Error(codes.NotFound, fmt.Sprintf("failed to find VM with domain.firmware.uuid %v", nodeID))
 }
+
+func (c *ControllerService) getAllowedZoneAndRegion(_ context.Context, requirement *csi.TopologyRequirement) (string, string, error) {
+	var zone, region string
+
+	if requirement == nil {
+		return "", "", nil
+	}
+
+	for _, topology := range requirement.GetPreferred() {
+		preferredZone, exists := topology.GetSegments()[WellKnownZoneTopologyKey]
+		if exists {
+			zone = preferredZone
+		}
+
+		preferredRegion, exists := topology.GetSegments()[WellKnownRegionTopologyKey]
+		if exists {
+			region = preferredRegion
+		}
+	}
+
+	for _, topology := range requirement.GetRequisite() {
+		requisiteZone, exists := topology.GetSegments()[WellKnownZoneTopologyKey]
+		if exists {
+			zone = requisiteZone
+		}
+
+		requisiteRegion, exists := topology.GetSegments()[WellKnownRegionTopologyKey]
+		if exists {
+			region = requisiteRegion
+		}
+	}
+
+	return zone, region, nil
+
+}

pkg/service/driver.go

@@ -8,6 +8,11 @@ import (
 	"kubevirt.io/csi-driver/pkg/util"
 )
 
+const (
+	WellKnownRegionTopologyKey = "topology.kubernetes.io/region"
+	WellKnownZoneTopologyKey   = "topology.kubernetes.io/zone"
+)
+
 var (
 	// VendorVersion is the vendor version set by ldflags at build time
 	VendorVersion = "0.2.0"
@@ -30,7 +35,8 @@ func NewKubevirtCSIDriver(virtClient kubevirt.Client,
 	storageClassEnforcement util.StorageClassEnforcement,
 	nodeID string,
 	runNodeService bool,
-	runControllerService bool) *KubevirtCSIDriver {
+	runControllerService bool,
+	allowedTopologies map[string]string) *KubevirtCSIDriver {
 	d := KubevirtCSIDriver{
 		IdentityService: NewIdentityService(identityClientset),
 	}
@@ -45,7 +51,7 @@ func NewKubevirtCSIDriver(virtClient kubevirt.Client,
 	}
 
 	if runNodeService {
-		d.NodeService = NewNodeService(nodeID)
+		d.NodeService = NewNodeService(nodeID, allowedTopologies)
 	}
 
 	return &d

pkg/service/identity.go

@@ -55,6 +55,13 @@ func (i *IdentityService) GetPluginCapabilities(context.Context, *csi.GetPluginC
 					},
 				},
 			},
+			{
+				Type: &csi.PluginCapability_Service_{
+					Service: &csi.PluginCapability_Service{
+						Type: csi.PluginCapability_Service_VOLUME_ACCESSIBILITY_CONSTRAINTS,
+					},
+				},
+			},
 		},
 	}, nil
 }

pkg/service/node.go

@@ -26,11 +26,12 @@ var nodeCaps = []csi.NodeServiceCapability_RPC_Type{
 // NodeService implements the CSI Driver node service
 type NodeService struct {
 	csi.UnimplementedNodeServer
-	nodeID       string
-	deviceLister DeviceLister
-	fsMaker      FsMaker
-	mounter      mount.Interface
-	dirMaker     dirMaker
+	nodeID            string
+	deviceLister      DeviceLister
+	fsMaker           FsMaker
+	mounter           mount.Interface
+	dirMaker          dirMaker
+	allowedTopologies map[string]string
 }
 
 type DeviceLister interface{ List() ([]byte, error) }
@@ -59,12 +60,13 @@ var NewFsMaker = func() FsMaker {
 	})
 }
 
-func NewNodeService(nodeId string) *NodeService {
+func NewNodeService(nodeId string, allowedTopologies map[string]string) *NodeService {
 	return &NodeService{
-		nodeID:       nodeId,
-		deviceLister: NewDeviceLister(),
-		fsMaker:      NewFsMaker(),
-		mounter:      NewMounter(),
+		nodeID:            nodeId,
+		deviceLister:      NewDeviceLister(),
+		fsMaker:           NewFsMaker(),
+		mounter:           NewMounter(),
+		allowedTopologies: allowedTopologies,
 		dirMaker: dirMakerFunc(func(path string, perm os.FileMode) error {
 			// MkdirAll returns nil if path already exists
 			return os.MkdirAll(path, perm)
@@ -334,7 +336,10 @@ func (n *NodeService) NodeExpandVolume(context.Context, *csi.NodeExpandVolumeReq
 // NodeGetInfo returns the node ID
 func (n *NodeService) NodeGetInfo(context.Context, *csi.NodeGetInfoRequest) (*csi.NodeGetInfoResponse, error) {
 	// the nodeID is the VM's ID in kubevirt or VMI.spec.domain.firmware.uuid
-	return &csi.NodeGetInfoResponse{NodeId: n.nodeID}, nil
+
+	return &csi.NodeGetInfoResponse{
+		NodeId:             n.nodeID,
+		AccessibleTopology: &csi.Topology{Segments: n.allowedTopologies}}, nil
 }
 
 // NodeGetCapabilities returns the supported capabilities of the node service