feat: add NVMe-oF connect concurrency limiter (#131)

Prevent kernel NVMe subsystem registration lock contention when staging
many volumes simultaneously on a single node. Uses a channel-based
semaphore (default: 5) configurable via --max-concurrent-nvme-connects
flag and node.maxConcurrentNVMeConnects Helm value. Adds Prometheus
metrics for concurrent/waiting connect operations.

Co-authored-by: Ashton Kinslow <github@ashtonkinslow.com>

Author: fenio

README.md

@@ -205,6 +205,7 @@ The driver is configured via command-line flags and Kubernetes secrets:
 - `--driver-name` - CSI driver name (default: `tns.csi.io`)
 - `--api-url` - TrueNAS API URL (e.g., `ws://YOUR-TRUENAS-IP/api/v2.0/websocket`)
 - `--api-key` - TrueNAS API key
+- `--max-concurrent-nvme-connects` - Maximum concurrent NVMe-oF connect operations per node (default: `5`)
 
 ### Storage Class Parameters
 

charts/tns-csi-driver/README.md

@@ -286,6 +286,7 @@ Detached snapshots use `zfs send/receive` to create independent dataset copies t
 | `node.kubeletPath` | Kubelet data directory | `/var/lib/kubelet` |
 | `node.logLevel` | Log verbosity (0-5) | `2` |
 | `node.debug` | Enable debug mode | `false` |
+| `node.maxConcurrentNVMeConnects` | Max concurrent NVMe-oF connect operations per node | `5` |
 | `node.resources.limits.cpu` | CPU limit | `200m` |
 | `node.resources.limits.memory` | Memory limit | `200Mi` |
 | `node.resources.requests.cpu` | CPU request | `10m` |

charts/tns-csi-driver/templates/node.yaml

@@ -58,6 +58,7 @@ spec:
             {{- if .Values.node.enableNVMeDiscovery }}
             - "--enable-nvme-discovery"
             {{- end }}
+            - "--max-concurrent-nvme-connects={{ .Values.node.maxConcurrentNVMeConnects | default 5 }}"
           env:
             - name: NODE_ID
               valueFrom:

charts/tns-csi-driver/values.yaml

@@ -150,6 +150,11 @@ node:
   # Enable only if you need discovery for multi-path or dynamic target topologies.
   enableNVMeDiscovery: false
 
+  # Maximum concurrent NVMe-oF connect operations per node.
+  # Limits contention on the kernel's NVMe subsystem registration lock.
+  # Recommended: 3-5. Set to 0 for unlimited (not recommended with >10 volumes per node).
+  maxConcurrentNVMeConnects: 5
+
   # Update strategy for DaemonSet
   updateStrategy:
     type: RollingUpdate

cmd/tns-csi-driver/main.go

@@ -20,16 +20,17 @@ var (
 )
 
 var (
-	endpoint      = flag.String("endpoint", "unix:///var/lib/kubelet/plugins/tns.csi.io/csi.sock", "CSI endpoint")
-	nodeID        = flag.String("node-id", "", "Node ID")
-	driverName    = flag.String("driver-name", "tns.csi.io", "Name of the driver")
-	apiURL        = flag.String("api-url", "", "Storage system API URL (e.g., ws://10.10.20.100/api/v2.0/websocket)")
-	apiKey        = flag.String("api-key", "", "Storage system API key")
-	metricsAddr   = flag.String("metrics-addr", ":8080", "Address to expose Prometheus metrics")
-	skipTLSVerify = flag.Bool("skip-tls-verify", false, "Skip TLS certificate verification (for self-signed certificates)")
-	showVersion         = flag.Bool("show-version", false, "Show version and exit")
-	debug               = flag.Bool("debug", false, "Enable debug logging (equivalent to -v=4)")
-	enableNVMeDiscovery = flag.Bool("enable-nvme-discovery", false, "Run nvme discover before nvme connect (default: false, all connection params are known from volume context)")
+	endpoint                  = flag.String("endpoint", "unix:///var/lib/kubelet/plugins/tns.csi.io/csi.sock", "CSI endpoint")
+	nodeID                    = flag.String("node-id", "", "Node ID")
+	driverName                = flag.String("driver-name", "tns.csi.io", "Name of the driver")
+	apiURL                    = flag.String("api-url", "", "Storage system API URL (e.g., ws://10.10.20.100/api/v2.0/websocket)")
+	apiKey                    = flag.String("api-key", "", "Storage system API key")
+	metricsAddr               = flag.String("metrics-addr", ":8080", "Address to expose Prometheus metrics")
+	skipTLSVerify             = flag.Bool("skip-tls-verify", false, "Skip TLS certificate verification (for self-signed certificates)")
+	showVersion               = flag.Bool("show-version", false, "Show version and exit")
+	debug                     = flag.Bool("debug", false, "Enable debug logging (equivalent to -v=4)")
+	enableNVMeDiscovery       = flag.Bool("enable-nvme-discovery", false, "Run nvme discover before nvme connect (default: false, all connection params are known from volume context)")
+	maxConcurrentNVMeConnects = flag.Int("max-concurrent-nvme-connects", 5, "Maximum number of concurrent NVMe-oF connect operations per node (limits kernel NVMe subsystem lock contention)")
 )
 
 func main() {
@@ -72,15 +73,16 @@ func main() {
 	klog.V(4).Infof("Node ID: %s", *nodeID)
 
 	drv, err := driver.NewDriver(driver.Config{
-		DriverName:          *driverName,
-		Version:             version,
-		NodeID:              *nodeID,
-		Endpoint:            *endpoint,
-		APIURL:              *apiURL,
-		APIKey:              *apiKey,
-		MetricsAddr:         *metricsAddr,
-		SkipTLSVerify:       *skipTLSVerify,
-		EnableNVMeDiscovery: *enableNVMeDiscovery,
+		DriverName:                *driverName,
+		Version:                   version,
+		NodeID:                    *nodeID,
+		Endpoint:                  *endpoint,
+		APIURL:                    *apiURL,
+		APIKey:                    *apiKey,
+		MetricsAddr:               *metricsAddr,
+		SkipTLSVerify:             *skipTLSVerify,
+		EnableNVMeDiscovery:       *enableNVMeDiscovery,
+		MaxConcurrentNVMeConnects: *maxConcurrentNVMeConnects,
 	})
 	if err != nil {
 		klog.Fatalf("Failed to create driver: %v", err)

docs/DEPLOYMENT.md

@@ -617,15 +617,20 @@ kubectl logs -n kube-system tns-csi-node-xxxxx -c tns-csi-plugin
    - Verify subsystem exists: `sudo nvme list-subsys`
    - Check /sys/class/nvme for device entries
 
-7. **iSCSI connection failures**
+7. **NVMe-oF volumes timing out with many concurrent mounts**
+   - Symptom: `signal: killed` in node plugin logs when staging many NVMe-oF volumes simultaneously
+   - Cause: Too many concurrent `nvme connect` processes overwhelming the kernel's NVMe subsystem registration lock
+   - Fix: The driver limits concurrency to 5 by default (`node.maxConcurrentNVMeConnects`). Lower this value if you still see timeouts, or increase it if mounts are too slow on fast hardware
+
+8. **iSCSI connection failures**
    - Verify open-iscsi is installed: `iscsiadm --version`
    - Check iscsid service is running: `systemctl status iscsid`
    - Verify iSCSI service is enabled on TrueNAS
    - Check firewall allows port 3260 (default iSCSI port)
    - Test discovery: `sudo iscsiadm -m discovery -t sendtargets -p YOUR-TRUENAS-IP:3260`
    - Check node plugin logs for detailed error messages
 
-8. **iSCSI device not appearing**
+9. **iSCSI device not appearing**
    - Wait a few seconds for device discovery
    - Check dmesg for SCSI errors: `sudo dmesg | grep -i scsi`
    - List active sessions: `sudo iscsiadm -m session`

docs/METRICS.md

@@ -38,6 +38,15 @@ Protocol-specific volume operations (NFS, NVMe-oF, and iSCSI):
   - Capacity of provisioned volumes in bytes
   - Labels: `volume_id`, `protocol`
 
+### NVMe-oF Connect Concurrency Metrics
+
+- **`tns_csi_nvme_connect_concurrent`** (gauge)
+  - Number of NVMe-oF connect operations currently in progress
+
+- **`tns_csi_nvme_connect_waiting`** (gauge)
+  - Number of NVMe-oF connect operations waiting for the semaphore
+  - Non-zero values indicate the concurrency limit is actively throttling connections
+
 ### WebSocket Connection Metrics
 
 Metrics for the TrueNAS API WebSocket connection:

pkg/driver/driver.go

@@ -20,16 +20,17 @@ import (
 
 // Config contains the configuration for the driver.
 type Config struct {
-	DriverName          string
-	Version             string
-	NodeID              string
-	Endpoint            string
-	APIURL              string
-	APIKey              string
-	MetricsAddr         string // Address to expose Prometheus metrics (e.g., ":8080")
-	TestMode            bool   // Enable test mode for sanity tests (skips actual mounts)
-	SkipTLSVerify       bool   // Skip TLS certificate verification (for self-signed certs)
-	EnableNVMeDiscovery bool   // Run nvme discover before nvme connect (default: false)
+	DriverName                string
+	Version                   string
+	NodeID                    string
+	Endpoint                  string
+	APIURL                    string
+	APIKey                    string
+	MetricsAddr               string // Address to expose Prometheus metrics (e.g., ":8080")
+	TestMode                  bool   // Enable test mode for sanity tests (skips actual mounts)
+	SkipTLSVerify             bool   // Skip TLS certificate verification (for self-signed certs)
+	EnableNVMeDiscovery       bool   // Run nvme discover before nvme connect (default: false)
+	MaxConcurrentNVMeConnects int    // Max concurrent NVMe-oF connect operations per node (default: 5)
 }
 
 // Driver is the TNS CSI driver.
@@ -75,7 +76,7 @@ func NewDriverWithClient(cfg Config, client tnsapi.ClientInterface) (*Driver, er
 	// Initialize CSI services
 	d.identity = NewIdentityService(cfg.DriverName, cfg.Version)
 	d.controller = NewControllerService(client, nodeRegistry)
-	d.node = NewNodeService(cfg.NodeID, client, cfg.TestMode, nodeRegistry, cfg.EnableNVMeDiscovery)
+	d.node = NewNodeService(cfg.NodeID, client, cfg.TestMode, nodeRegistry, cfg.EnableNVMeDiscovery, cfg.MaxConcurrentNVMeConnects)
 
 	return d, nil
 }

pkg/driver/node.go

@@ -37,19 +37,24 @@ type NodeService struct {
 	csi.UnimplementedNodeServer
 	apiClient       tnsapi.ClientInterface
 	nodeRegistry    *NodeRegistry
+	nvmeConnectSem  chan struct{}
 	nodeID          string
-	testMode        bool // Test mode flag to skip actual mounts
-	enableDiscovery bool // Run nvme discover before nvme connect
+	testMode        bool
+	enableDiscovery bool
 }
 
 // NewNodeService creates a new node service.
-func NewNodeService(nodeID string, apiClient tnsapi.ClientInterface, testMode bool, nodeRegistry *NodeRegistry, enableDiscovery bool) *NodeService {
+func NewNodeService(nodeID string, apiClient tnsapi.ClientInterface, testMode bool, nodeRegistry *NodeRegistry, enableDiscovery bool, maxConcurrentNVMeConnects int) *NodeService {
+	if maxConcurrentNVMeConnects <= 0 {
+		maxConcurrentNVMeConnects = 5
+	}
 	return &NodeService{
 		nodeID:          nodeID,
 		apiClient:       apiClient,
 		testMode:        testMode,
 		nodeRegistry:    nodeRegistry,
 		enableDiscovery: enableDiscovery,
+		nvmeConnectSem:  make(chan struct{}, maxConcurrentNVMeConnects),
 	}
 }
 

pkg/driver/node_nvmeof.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/fenio/tns-csi/pkg/metrics"
 	"github.com/fenio/tns-csi/pkg/mount"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -78,6 +79,27 @@ func (s *NodeService) stageNVMeOFVolume(ctx context.Context, req *csi.NodeStageV
 		return nil, status.Errorf(codes.FailedPrecondition, "nvme-cli not available: %v", checkErr)
 	}
 
+	// Acquire semaphore to limit concurrent NVMe-oF connect operations.
+	// This prevents overwhelming the kernel's NVMe subsystem registration lock
+	// when many volumes are being staged simultaneously.
+	klog.V(4).Infof("Waiting for NVMe-oF connect semaphore (capacity: %d) for NQN: %s", cap(s.nvmeConnectSem), params.nqn)
+	metrics.NVMeConnectWaiting()
+	select {
+	case s.nvmeConnectSem <- struct{}{}:
+		metrics.NVMeConnectDoneWaiting()
+		metrics.NVMeConnectStart()
+		defer func() {
+			<-s.nvmeConnectSem
+			metrics.NVMeConnectDone()
+		}()
+	case <-ctx.Done():
+		metrics.NVMeConnectDoneWaiting()
+		return nil, status.Errorf(codes.DeadlineExceeded,
+			"timed out waiting for NVMe-oF connect semaphore (max concurrent: %d): %v",
+			cap(s.nvmeConnectSem), ctx.Err())
+	}
+	klog.V(4).Infof("Acquired NVMe-oF connect semaphore for NQN: %s", params.nqn)
+
 	// Connect to NVMe-oF target and stage device
 	return s.connectAndStageDevice(ctx, params, volumeID, stagingTargetPath, volumeCapability, isBlockVolume, volumeContext, datasetName)
 }