nvmeof: add external client support via allowHostNQNs parameter

gadididi · gadididi · commit ba8d81124fd0 · 2026-04-27T10:11:26.000+03:00
Implements dynamic host access control for NVMe-oF volumes to support
external(non-Kubernetes) clients through VolumeAttributesClass.

Changes:
- Add `parseHostsParameters()` to parse YAML host list from VAC
- Implement `modifyNVMeoFHosts()` for runtime host updates
- Add `ListHosts()` and `UpdateHostsForSubsystem()` for host updateing
- Support allowHostNQNs parameter in CreateVolume and ControllerModifyVolume
- Auto-generate subsystem NQN from volumeID if not provided

Users can now specify external hosts in VAC mutable parameters:
  allowHostNQNs: |
    - nqn.2014-08.org.nvmexpress:host1
    - nqn.2014-08.org.nvmexpress:host2

Signed-off-by: gadi-didi &lt;gadi.didi@ibm.com&gt;
diff --git a/internal/nvmeof/controller/controllerserver.go b/internal/nvmeof/controller/controllerserver.go
@@ -25,6 +25,7 @@ import (
 	"strconv"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/ghodss/yaml"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
@@ -166,7 +167,7 @@ func (cs *Server) CreateVolume(
 		}
 	}()
 
-	nvmeofData, err = cs.createNVMeoFResources(ctx, req, rbdPoolName, rbdRadosNameSpace, rbdImageName)
+	nvmeofData, err = cs.createNVMeoFResources(ctx, req, rbdPoolName, rbdRadosNameSpace, rbdImageName, volumeID)
 	if err != nil {
 		log.ErrorLog(ctx, "NVMe-oF resource setup failed for volumeID %s: %v", volumeID, err)
 
@@ -352,11 +353,22 @@ func (cs *Server) ControllerModifyVolume(
 
 		return nil, status.Errorf(codes.InvalidArgument, "failed to parse QoS parameters: %v", err)
 	}
+	hostsList, err := parseHostsParameters(params)
+	if err != nil {
+		log.ErrorLog(ctx, "failed to parse NVMe-oF hosts parameters: %v", err)
+
+		return nil, status.Errorf(codes.InvalidArgument, "failed to parse hosts parameters: %v", err)
+	}
 	if nvmeofQoS != nil {
 		if err := cs.modifyNVMeoFQoS(ctx, req, nvmeofQoS); err != nil {
 			return nil, err
 		}
 	}
+	if hostsList != nil {
+		if err := cs.modifyNVMeoFHosts(ctx, req, hostsList); err != nil {
+			return nil, err
+		}
+	}
 
 	return &csi.ControllerModifyVolumeResponse{}, nil
 }
@@ -466,6 +478,11 @@ func validateCreateVolumeRequest(req *csi.CreateVolumeRequest) error {
 	if err != nil {
 		return fmt.Errorf("invalid NVMe-oF QoS parameters: %w", err)
 	}
+
+	_, err = parseHostsParameters(mutableParams)
+	if err != nil {
+		return fmt.Errorf("invalid NVMe-oF hosts parameters (for external clients): %w", err)
+	}
 	err = validateDHCHAPParameter(params["dhchapMode"])
 	if err != nil {
 		return err
@@ -532,42 +549,22 @@ func validateNetworkMask(networkMask string) error {
 	return nil
 }
 
-// parseQoSParameters extracts and parses QoS parameters from the given map.
-func parseQoSParameters(params map[string]string) (*nvmeof.NVMeoFQosVolume, error) {
-	qos := &nvmeof.NVMeoFQosVolume{}
-	hasAnyQoS := false
-
-	parseParam := func(key, name string, dest **uint64) error {
-		if val, exists := params[key]; exists && val != "" {
-			parsed, err := strconv.ParseUint(val, 10, 64)
-			if err != nil {
-				return fmt.Errorf("invalid %s: %w", name, err)
-			}
-			*dest = &parsed
-			hasAnyQoS = true
-		}
-
-		return nil
-	}
-
-	if err := parseParam(nvmeof.RwIosPerSecond, nvmeof.RwIosPerSecond, &qos.RwIosPerSecond); err != nil {
-		return nil, err
-	}
-	if err := parseParam(nvmeof.RwMbytesPerSecond, nvmeof.RwMbytesPerSecond, &qos.RwMbytesPerSecond); err != nil {
-		return nil, err
-	}
-	if err := parseParam(nvmeof.RMbytesPerSecond, nvmeof.RMbytesPerSecond, &qos.RMbytesPerSecond); err != nil {
-		return nil, err
+// parseHostsParameters parses the hosts yaml list parameter and validates its contents.
+// It returns a slice of hostnames or an error if the YAML is invalid.
+func parseHostsParameters(params map[string]string) ([]string, error) {
+	allowHostNQNs, exists := params[AllowHostNQNs]
+	if !exists || allowHostNQNs == "" {
+		return nil, nil
 	}
-	if err := parseParam(nvmeof.WMbytesPerSecond, nvmeof.WMbytesPerSecond, &qos.WMbytesPerSecond); err != nil {
-		return nil, err
+	var allowHostsList []string
+	if err := yaml.Unmarshal([]byte(allowHostNQNs), &allowHostsList); err != nil {
+		return nil, fmt.Errorf("invalid %s: must be a YAML list of strings: %w", AllowHostNQNs, err)
 	}
-
-	if !hasAnyQoS {
+	if len(allowHostsList) == 0 {
 		return nil, nil
 	}
 
-	return qos, nil
+	return allowHostsList, nil
 }
 
 // withGatewayConnection is a helper that manages the common pattern of:
@@ -628,6 +625,74 @@ func (cs *Server) withGatewayConnection(
 	return fn(ctx, gateway, nvmeofData)
 }
 
+// modifyNVMeoFHosts handles adding or removing hosts from the subsystem based on the provided list of host NQNs.
+func (cs *Server) modifyNVMeoFHosts(ctx context.Context, req *csi.ControllerModifyVolumeRequest, hosts []string) error {
+	volumeID := req.GetVolumeId()
+	if len(hosts) == 0 {
+		log.DebugLog(ctx, "No hosts to add or remove for volume %s", volumeID)
+
+		return nil
+	}
+
+	return cs.withGatewayConnection(ctx, req, volumeID, func(
+		ctx context.Context,
+		gateway *nvmeof.GatewayRpcClient,
+		nvmeofData *nvmeof.NVMeoFVolumeData,
+	) error {
+		log.DebugLog(ctx, "Modifying hosts for subsystem=%s, nsid=%d: desired hosts=%v",
+			nvmeofData.SubsystemNQN, nvmeofData.NamespaceID, hosts)
+
+		err := gateway.UpdateHostsForSubsystem(ctx, nvmeofData.SubsystemNQN, hosts)
+		if err != nil {
+			log.ErrorLog(ctx, "Failed to update hosts for subsystem: %v", err)
+
+			return status.Errorf(codes.Internal, "failed to update hosts for subsystem: %v", err)
+		}
+
+		log.DebugLog(ctx, "Successfully modified hosts for volume %s", volumeID)
+
+		return nil
+	})
+}
+
+// parseQoSParameters extracts and parses QoS parameters from the given map.
+func parseQoSParameters(params map[string]string) (*nvmeof.NVMeoFQosVolume, error) {
+	qos := &nvmeof.NVMeoFQosVolume{}
+	hasAnyQoS := false
+
+	parseParam := func(key, name string, dest **uint64) error {
+		if val, exists := params[key]; exists && val != "" {
+			parsed, err := strconv.ParseUint(val, 10, 64)
+			if err != nil {
+				return fmt.Errorf("invalid %s: %w", name, err)
+			}
+			*dest = &parsed
+			hasAnyQoS = true
+		}
+
+		return nil
+	}
+
+	if err := parseParam(nvmeof.RwIosPerSecond, nvmeof.RwIosPerSecond, &qos.RwIosPerSecond); err != nil {
+		return nil, err
+	}
+	if err := parseParam(nvmeof.RwMbytesPerSecond, nvmeof.RwMbytesPerSecond, &qos.RwMbytesPerSecond); err != nil {
+		return nil, err
+	}
+	if err := parseParam(nvmeof.RMbytesPerSecond, nvmeof.RMbytesPerSecond, &qos.RMbytesPerSecond); err != nil {
+		return nil, err
+	}
+	if err := parseParam(nvmeof.WMbytesPerSecond, nvmeof.WMbytesPerSecond, &qos.WMbytesPerSecond); err != nil {
+		return nil, err
+	}
+
+	if !hasAnyQoS {
+		return nil, nil
+	}
+
+	return qos, nil
+}
+
 // modifyNVMeoFQoS handles NVMe-oF gateway QoS modification.
 func (cs *Server) modifyNVMeoFQoS(
 	ctx context.Context,
@@ -759,15 +824,16 @@ func (cs *Server) createNVMeoFResources(
 	req *csi.CreateVolumeRequest,
 	rbdPoolName,
 	rbdRadosNameSpace,
-	rbdImageName string,
+	rbdImageName,
+	volumeID string,
 ) (*nvmeof.NVMeoFVolumeData, error) {
 	// Step 1: Extract parameters (already validated)
 	params := req.GetParameters()
 
 	networkMask := params["networkMask"]
 	nvmeofData := &nvmeof.NVMeoFVolumeData{}
 
-	if err := nvmeofData.SetFromParameters(params); err != nil {
+	if err := nvmeofData.SetFromParameters(params, volumeID); err != nil {
 		return nil, fmt.Errorf("failed to set NVMe-oF volume data: %w", err)
 	}
 
@@ -781,7 +847,14 @@ func (cs *Server) createNVMeoFResources(
 
 		return nil, fmt.Errorf("failed to parse QoS parameters: %w", err)
 	}
+	// If VAC with hosts list is given (for external client)
+	// We need to parse the hosts list and pass it to the gateway for creating host entries and adding them to the subsystem.
+	hosts, err := parseHostsParameters(mutableParams)
+	if err != nil {
+		log.ErrorLog(ctx, "failed to parse NVMe-oF hosts parameters: %v", err)
 
+		return nil, fmt.Errorf("failed to parse hosts parameters: %w", err)
+	}
 	// Step 2: Connect to gateway
 	config, err := getGatewayConfigFromRequest(params)
 	if err != nil {
@@ -829,7 +902,16 @@ func (cs *Server) createNVMeoFResources(
 			return nvmeofData, fmt.Errorf("setting QoS limits failed: %w", err)
 		}
 	}
-
+	if hosts != nil {
+		log.DebugLog(ctx, "Adding hosts to subsystem: %v", hosts)
+		for _, host := range hosts {
+			// TODO - for now we create host with empty DH-CHAP keys,
+			// in the future we can extend the VAC parameters to allow passing DH-CHAP keys for each host if needed??
+			if err := gateway.AddHost(ctx, nvmeofData.SubsystemNQN, host, nvmeof.DHCHAPKeys{}); err != nil {
+				return nvmeofData, fmt.Errorf("adding host %s to subsystem failed: %w", host, err)
+			}
+		}
+	}
 	// Step 6: If using auto-listeners, query them back for storing in metadata
 	if networkMask != "" {
 		autoListeners, err := gateway.ListListeners(ctx, nvmeofData.SubsystemNQN)
@@ -1029,6 +1111,15 @@ func getHostNQNFromNodeID(nodeID string) (string, error) {
 	return prefix + nodeID, nil
 }
 
+// AllowHostNQNs is the VolumeAttributesClass mutable parameter key for specifying
+// a YAML list of host NQNs to allow access to a volume. Use "*" to allow any host.
+// Example:
+//
+//	allowHostNQNs: |
+//	  - nqn.2014-08.org.nvmexpress:host1
+//	  - nqn.2014-08.org.nvmexpress:host2
+const AllowHostNQNs = "allowHostNQNs"
+
 // VolumeContext metadata keys.
 const (
 	// NVMe-oF resource info.
diff --git a/internal/nvmeof/nvmeof.go b/internal/nvmeof/nvmeof.go
@@ -605,6 +605,107 @@ func (gw *GatewayRpcClient) RemoveHost(ctx context.Context, subsystemNQN, hostNQ
 	}
 }
 
+// ListHosts lists all hosts in a subsystem.
+// Returns a slice of host NQNs that have access to the subsystem.
+func (gw *GatewayRpcClient) ListHosts(ctx context.Context, subsystemNQN string) ([]string, error) {
+	log.DebugLog(ctx, "Listing hosts in subsystem %s", subsystemNQN)
+
+	req := &pb.ListHostsReq{
+		Subsystem: subsystemNQN,
+	}
+
+	resp, err := gw.client.ListHosts(ctx, req)
+	if err != nil {
+		return nil, fmt.Errorf("failed to list hosts in subsystem %s: %w", subsystemNQN, err)
+	}
+	if resp.GetStatus() != 0 {
+		return nil, fmt.Errorf("gateway ListHosts returned error (status=%d): %s",
+			resp.GetStatus(), resp.GetErrorMessage())
+	}
+
+	// Extract host NQNs from response
+	hosts := make([]string, 0, len(resp.GetHosts()))
+	for _, host := range resp.GetHosts() {
+		hosts = append(hosts, host.GetNqn())
+	}
+
+	log.DebugLog(ctx, "Listed %d hosts in subsystem %s", len(hosts), subsystemNQN)
+
+	return hosts, nil
+}
+
+// UpdateHostsForSubsystem reconciles the hosts in a subsystem to match the desired list.
+// It lists current hosts, then adds/removes hosts to ensure the subsystem has exactly
+// the hosts specified in desiredHosts.
+func (gw *GatewayRpcClient) UpdateHostsForSubsystem(
+	ctx context.Context,
+	subsystemNQN string,
+	desiredHosts []string,
+) error {
+	log.DebugLog(ctx, "Reconciling hosts for subsystem %s: desired=%v", subsystemNQN, desiredHosts)
+
+	// Get current hosts in the subsystem
+	currentHosts, err := gw.ListHosts(ctx, subsystemNQN)
+	if err != nil {
+		return fmt.Errorf("failed to list current hosts: %w", err)
+	}
+
+	// Convert to sets for easier comparison
+	currentHostSet := make(map[string]bool)
+	for _, host := range currentHosts {
+		currentHostSet[host] = true
+	}
+
+	desiredHostSet := make(map[string]bool)
+	for _, host := range desiredHosts {
+		desiredHostSet[host] = true
+	}
+
+	// Determine hosts to add and remove
+	var hostsToAdd []string
+	var hostsToRemove []string
+
+	// Find hosts that need to be added
+	for _, host := range desiredHosts {
+		if !currentHostSet[host] {
+			hostsToAdd = append(hostsToAdd, host)
+		}
+	}
+
+	// Find hosts that need to be removed
+	for _, host := range currentHosts {
+		if !desiredHostSet[host] {
+			hostsToRemove = append(hostsToRemove, host)
+		}
+	}
+
+	log.DebugLog(ctx, "Host reconciliation for subsystem %s: current=%v, desired=%v, toAdd=%v, toRemove=%v",
+		subsystemNQN, currentHosts, desiredHosts, hostsToAdd, hostsToRemove)
+
+	// Remove hosts that shouldn't be there
+	for _, host := range hostsToRemove {
+		log.DebugLog(ctx, "Removing host %s from subsystem %s", host, subsystemNQN)
+		if err := gw.RemoveHost(ctx, subsystemNQN, host); err != nil {
+			return fmt.Errorf("failed to remove host %s: %w", host, err)
+		}
+	}
+
+	// Add hosts that should be there
+	for _, host := range hostsToAdd {
+		log.DebugLog(ctx, "Adding host %s to subsystem %s", host, subsystemNQN)
+		// Note: AddHost requires DHCHAPKeys, but for now we pass empty keys
+		// TODO: Support DH-CHAP keys if needed for host updates
+		if err := gw.AddHost(ctx, subsystemNQN, host, DHCHAPKeys{}); err != nil {
+			return fmt.Errorf("failed to add host %s: %w", host, err)
+		}
+	}
+
+	log.DebugLog(ctx, "Successfully reconciled hosts for subsystem %s: added %d, removed %d",
+		subsystemNQN, len(hostsToAdd), len(hostsToRemove))
+
+	return nil
+}
+
 // List namespaces in a subsystem.
 func (gw *GatewayRpcClient) ListNamespaces(ctx context.Context, subsystemNQN string) (*pb.NamespacesInfo, error) {
 	log.DebugLog(ctx, "Listing namespaces in subsystem %s", subsystemNQN)
diff --git a/internal/nvmeof/volume.go b/internal/nvmeof/volume.go
@@ -84,10 +84,12 @@ func SetupListeners(listenersJSON string) ([]ListenerDetails, error) {
 // It extracts the subsystem NQN, gateway management info, security config, and
 // listener info from the parameters.
 // It also applies default values to listeners if necessary.
-func (v *NVMeoFVolumeData) SetFromParameters(parameters map[string]string) error {
+func (v *NVMeoFVolumeData) SetFromParameters(parameters map[string]string, volumeID string) error {
 	// set subsystem NQN
 	v.SubsystemNQN = parameters["subsystemNQN"]
-
+	if v.SubsystemNQN == "" {
+		v.SubsystemNQN = fmt.Sprintf("nqn.2016-06.io.ceph:subsystem.%s", volumeID)
+	}
 	// set gw management info
 	if nvmeofGatewayPortStr := parameters["nvmeofGatewayPort"]; nvmeofGatewayPortStr != "" {
 		parsed, err := strconv.ParseUint(nvmeofGatewayPortStr, 10, 32)
