Introduced parallelism for NAS Driver in node part of the Trident codebase

Author: shashank-netapp

frontend/csi/node_server.go

@@ -1047,9 +1047,9 @@ func TestFixISCSISessions(t *testing.T) {
 			portals := getPortals(input.PublishedPortals, input.PortalActions)
 
 			if input.AddNewNodeOps {
-				go utils.Lock(ctx, "test-lock1", lockID)
+				go utils.Lock(ctx, "test-lock1", nodeLockID)
 				snooze(10)
-				go utils.Lock(ctx, "test-lock2", lockID)
+				go utils.Lock(ctx, "test-lock2", nodeLockID)
 				snooze(10)
 			}
 
@@ -1074,16 +1074,16 @@ func TestFixISCSISessions(t *testing.T) {
 			}
 
 			if input.AddNewNodeOps {
-				utils.Unlock(ctx, "test-lock1", lockID)
+				utils.Unlock(ctx, "test-lock1", nodeLockID)
 
 				// Wait for the lock to be released
-				for utils.WaitQueueSize(lockID) > 1 {
+				for utils.WaitQueueSize(nodeLockID) > 1 {
 					snooze(10)
 				}
 
 				// Give some time for another context to acquire the lock
 				snooze(100)
-				utils.Unlock(ctx, "test-lock2", lockID)
+				utils.Unlock(ctx, "test-lock2", nodeLockID)
 			}
 		})
 	}
@@ -1599,23 +1599,23 @@ func TestAttemptLock_Failure(t *testing.T) {
 	lockContext := "fakeLockContext-req1"
 	lockTimeout := 200 * time.Millisecond
 	// first request takes the lock
-	expected := attemptLock(ctx, lockContext, lockTimeout)
+	expected := attemptLock(ctx, lockContext, nodeLockID, lockTimeout)
 
 	// start the second request so that it is in race for the lock
 	go func() {
 		defer wg.Done()
 		ctx := context.Background()
 		lockContext := "fakeLockContext-req2"
-		expected := attemptLock(ctx, lockContext, lockTimeout)
+		expected := attemptLock(ctx, lockContext, nodeLockID, lockTimeout)
 
 		assert.False(t, expected)
-		utils.Unlock(ctx, lockContext, lockID)
+		utils.Unlock(ctx, lockContext, nodeLockID)
 	}()
 	// first request goes to sleep holding the lock
 	if expected {
 		time.Sleep(500 * time.Millisecond)
 	}
-	utils.Unlock(ctx, lockContext, lockID)
+	utils.Unlock(ctx, lockContext, nodeLockID)
 	wg.Wait()
 }
 
@@ -1632,7 +1632,7 @@ func TestAttemptLock_Success(t *testing.T) {
 	lockContext := "fakeLockContext-req1"
 	lockTimeout := 500 * time.Millisecond
 	// first request takes the lock
-	expected := attemptLock(ctx, lockContext, lockTimeout)
+	expected := attemptLock(ctx, lockContext, nodeLockID, lockTimeout)
 
 	// start the second request so that it is in race for the lock
 	go func() {
@@ -1641,16 +1641,16 @@ func TestAttemptLock_Success(t *testing.T) {
 		lockContext := "fakeLockContext-req2"
 		lockTimeout := 5 * time.Second
 
-		expected := attemptLock(ctx, lockContext, lockTimeout)
+		expected := attemptLock(ctx, lockContext, nodeLockID, lockTimeout)
 
 		assert.True(t, expected)
-		utils.Unlock(ctx, lockContext, lockID)
+		utils.Unlock(ctx, lockContext, nodeLockID)
 	}()
 	// first request goes to sleep holding the lock
 	if expected {
 		time.Sleep(200 * time.Millisecond)
 	}
-	utils.Unlock(ctx, lockContext, lockID)
+	utils.Unlock(ctx, lockContext, nodeLockID)
 	wg.Wait()
 }
 

frontend/csi/node_server_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/netapp/trident/utils/fcp"
 	"github.com/netapp/trident/utils/filesystem"
 	"github.com/netapp/trident/utils/iscsi"
+	"github.com/netapp/trident/utils/limiter"
 	"github.com/netapp/trident/utils/models"
 	"github.com/netapp/trident/utils/mount"
 	"github.com/netapp/trident/utils/nvme"
@@ -73,6 +74,8 @@ type Plugin struct {
 
 	nodeIsRegistered bool
 
+	limiterSharedMap map[string]limiter.Limiter
+
 	iSCSISelfHealingTicker   *time.Ticker
 	iSCSISelfHealingChannel  chan struct{}
 	iSCSISelfHealingInterval time.Duration
@@ -196,6 +199,7 @@ func NewNodePlugin(
 		enableForceDetach:        enableForceDetach,
 		unsafeDetach:             unsafeDetach,
 		opCache:                  sync.Map{},
+		limiterSharedMap:         make(map[string]limiter.Limiter),
 		iSCSISelfHealingInterval: iSCSISelfHealingInterval,
 		iSCSISelfHealingWaitTime: iSCSIStaleSessionWaitTime,
 		nvmeHandler:              nvme.NewNVMeHandler(),
@@ -306,6 +310,7 @@ func NewAllInOnePlugin(
 		controllerHelper:         *controllerHelper,
 		nodeHelper:               *nodeHelper,
 		opCache:                  sync.Map{},
+		limiterSharedMap:         make(map[string]limiter.Limiter),
 		iSCSISelfHealingInterval: iSCSISelfHealingInterval,
 		iSCSISelfHealingWaitTime: iSCSIStaleSessionWaitTime,
 		nvmeHandler:              nvme.NewNVMeHandler(),
@@ -395,6 +400,9 @@ func (p *Plugin) Activate() error {
 			p.startISCSISelfHealingThread(ctx)
 			p.startNVMeSelfHealingThread(ctx)
 
+			// Initialize node scalability limiter.
+			p.InitializeNodeLimiter(ctx)
+
 			if p.enableForceDetach {
 				p.startReconcilingNodePublications(ctx)
 			}
@@ -618,3 +626,67 @@ func (p *Plugin) stopNVMeSelfHealingThread(_ context.Context) {
 
 	return
 }
+
+// InitializeNodeLimiter initializes the sharedMapLimiter of node with various csi workflow related limiters.
+// This function sets up limiters for different volume operations such as staging, unstaging,
+// publishing, and unpublishing NFS and SMB volumes. Each limiter is configured with a semaphoreN
+// to control the maximum number of concurrent operations allowed for each type of volume operation.
+func (p *Plugin) InitializeNodeLimiter(ctx context.Context) {
+	var err error
+
+	if p.limiterSharedMap[NodeStageNFSVolume], err = limiter.New(ctx,
+		NodeStageNFSVolume,
+		limiter.TypeSemaphoreN,
+		limiter.WithSemaphoreNSize(ctx, maxNodeStageNFSVolumeOperations),
+	); err != nil {
+		Logc(ctx).Fatalf("Failed to initialize limiter for %s: %v", NodeStageNFSVolume, err)
+	}
+
+	if p.limiterSharedMap[NodeStageSMBVolume], err = limiter.New(ctx,
+		NodeStageSMBVolume,
+		limiter.TypeSemaphoreN,
+		limiter.WithSemaphoreNSize(ctx, maxNodeStageSMBVolumeOperations),
+	); err != nil {
+		Logc(ctx).Fatalf("Failed to initialize limiter for %s: %v", NodeStageSMBVolume, err)
+	}
+
+	if p.limiterSharedMap[NodeUnstageNFSVolume], err = limiter.New(ctx,
+		NodeUnstageNFSVolume,
+		limiter.TypeSemaphoreN,
+		limiter.WithSemaphoreNSize(ctx, maxNodeUnstageNFSVolumeOperations),
+	); err != nil {
+		Logc(ctx).Fatalf("Failed to initialize limiter for %s: %v", NodeUnstageNFSVolume, err)
+	}
+
+	if p.limiterSharedMap[NodeUnstageSMBVolume], err = limiter.New(ctx,
+		NodeUnstageSMBVolume,
+		limiter.TypeSemaphoreN,
+		limiter.WithSemaphoreNSize(ctx, maxNodeUnstageSMBVolumeOperations),
+	); err != nil {
+		Logc(ctx).Fatalf("Failed to initialize limiter for %s: %v", NodeUnstageSMBVolume, err)
+	}
+
+	if p.limiterSharedMap[NodePublishNFSVolume], err = limiter.New(ctx,
+		NodePublishNFSVolume,
+		limiter.TypeSemaphoreN,
+		limiter.WithSemaphoreNSize(ctx, maxNodePublishNFSVolumeOperations),
+	); err != nil {
+		Logc(ctx).Fatalf("Failed to initialize limiter for %s: %v", NodePublishNFSVolume, err)
+	}
+
+	if p.limiterSharedMap[NodePublishSMBVolume], err = limiter.New(ctx,
+		NodePublishSMBVolume,
+		limiter.TypeSemaphoreN,
+		limiter.WithSemaphoreNSize(ctx, maxNodePublishSMBVolumeOperations),
+	); err != nil {
+		Logc(ctx).Fatalf("Failed to initialize limiter for %s: %v", NodePublishSMBVolume, err)
+	}
+
+	if p.limiterSharedMap[NodeUnpublishVolume], err = limiter.New(ctx,
+		NodeUnpublishVolume,
+		limiter.TypeSemaphoreN,
+		limiter.WithSemaphoreNSize(ctx, maxNodeUnpublishVolumeOperations),
+	); err != nil {
+		Logc(ctx).Fatalf("Failed to initialize limiter for %s: %v", NodeUnpublishVolume, err)
+	}
+}

frontend/csi/plugin.go

@@ -12,10 +12,13 @@ import (
 	"strings"
 
 	"github.com/spf13/afero"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
 	"github.com/netapp/trident/config"
 	"github.com/netapp/trident/internal/fiji"
 	. "github.com/netapp/trident/logging"
+	"github.com/netapp/trident/utils"
 	"github.com/netapp/trident/utils/errors"
 	"github.com/netapp/trident/utils/filesystem"
 	"github.com/netapp/trident/utils/models"
@@ -95,6 +98,13 @@ func (v *VolumePublishManager) WriteTrackingInfo(
 		return err
 	}
 
+	// we can have locks on filename itself.
+	lockContext := "WriteTrackingInfo"
+	defer utils.Unlock(ctx, lockContext, filename)
+	if !attemptLock(ctx, lockContext, filename, csiNodeLockTimeout) {
+		return status.Error(codes.Aborted, "request waited too long for the lock")
+	}
+
 	err := jsonRW.WriteJSONFile(ctx, trackingInfo, tmpFile, "volume tracking info")
 	if err != nil {
 		return err
@@ -133,6 +143,11 @@ func (v *VolumePublishManager) readTrackingInfo(
 		return nil, err
 	}
 
+	lockContext := "ReadTrackingInfo"
+	defer utils.Unlock(ctx, lockContext, filename)
+	if !attemptLock(ctx, lockContext, filename, csiNodeLockTimeout) {
+		return nil, status.Error(codes.Aborted, "request waited too long for the lock on tracking file")
+	}
 	err := jsonRW.ReadJSONFile(ctx, &volumeTrackingInfo, path.Join(v.volumeTrackingInfoPath, filename),
 		"volume tracking info")
 	if err != nil {

frontend/csi/volume_publish_manager.go

@@ -0,0 +1,59 @@
+package limiter
+
+//go:generate mockgen -destination=../../mocks/mock_utils/mock_limiter/mock_limiter.go -package=mock_limiter github.com/netapp/trident/utils/limiter Limiter
+
+import (
+	"context"
+	"fmt"
+
+	. "github.com/netapp/trident/logging"
+)
+
+type Limiter interface {
+	Wait(ctx context.Context) error
+	Release(ctx context.Context)
+}
+
+// LimiterOption Using an option pattern to generalize New function.
+type LimiterOption func(limiter Limiter) error
+
+type LimiterType uint16
+
+const (
+	TypeSemaphoreN LimiterType = iota
+)
+
+// New creates a limiter based on the provided limiterType.
+// Note: This function is not thread-safe, so precautions are necessary for concurrent use.
+// Ensure the limiter type and options are correctly matched to avoid errors.
+func New(ctx context.Context, limID string, limiterType LimiterType, options ...LimiterOption) (Limiter, error) {
+	logFields := LogFields{
+		"LimiterID":   limID,
+		"LimiterType": limiterType,
+		"Options":     options,
+	}
+	Logc(ctx).WithFields(logFields).Debug(">>>> Limiter.New")
+	defer Logc(ctx).WithFields(logFields).Debug("<<<< Limiter.New")
+
+	var limiter Limiter
+
+	// We'll create the new Limiter, depending upon the type passed in limiterType arg.
+	switch limiterType {
+	case TypeSemaphoreN:
+		limiter = newSemaphoreN(limID)
+
+	default:
+		return nil, fmt.Errorf("unknown limiter type: %T", limiterType)
+	}
+
+	// Applying the limiter options.
+	for _, option := range options {
+		if err := option(limiter); err != nil {
+			return nil, err
+		}
+	}
+
+	Logc(ctx).WithFields(logFields).Debugf("Limiter with limID %s was created", limID)
+
+	return limiter, nil
+}