Repository: kubernetes-csi/csi-driver-nfs (go)

Issue:
High number of threads on csi-nfs-node.
<!-- Please use this template while reporting a bug and provide as much info as possible. Not doing so may result in your bug not being addressed in a timely manner. Thanks!
-->


**What happened**:
High number of threads on csi-nfs-nodes.

**What you expected to happen**:
The thread count on de nodes remains normal.
 
**How to reproduce it**:

**Anything else we need to know?**:
The thread count on the on the csi-nfs-node pods keeps getting higher as time passes. This causes the system load to become higher as time passes because the number of threads just becomes higher and higer.
I dont know if the following messages are related but they might be:

On the csi-nfs-node pods the following messages are logged:
```
│ nfs I1219 15:22:14.747364       1 utils.go:107] GRPC call: /csi.v1.Node/NodeUnpublishVolume                                                                                                                                                           │
│ nfs I1219 15:22:14.747393       1 utils.go:108] GRPC request: {"target_path":"/var/lib/kubelet/pods/4f08f426-6185-4e2a-9ced-83ea62e4d739/volumes/kubernetes.io~csi/nfs-pv/mount","volume_id":"nfs-server-nfs-server-provisioner.nfs-server.svc.cluste │
│ nfs I1219 15:22:14.747476       1 nodeserver.go:157] NodeUnpublishVolume: unmounting volume nfs-server-nfs-server-provisioner.nfs-server.svc.cluster.local/export/pvc-084cfb0d-cb0a-4089-a6ce-5b29049f4347 on /var/lib/kubelet/pods/4f08f426-6185-4e2 │
│ nfs I1219 15:22:14.747484       1 nodeserver.go:162] force unmount nfs-server-nfs-server-provisioner.nfs-server.svc.cluster.local/export/pvc-084cfb0d-cb0a-4089-a6ce-5b29049f4347 on /var/lib/kubelet/pods/4f08f426-6185-4e2a-9ced-83ea62e4d739/volum │
│ nfs I1219 15:22:14.849002       1 utils.go:107] GRPC call: /csi.v1.Node/NodeUnpublishVolume                                                                                                                                                           │
│ nfs I1219 15:22:14.849029       1 utils.go:108] GRPC request: {"target_path":"/var/lib/kubelet/pods/bf38702d-92b6-40b4-bfc0-1c7ab46f1633/volumes/kubernetes.io~csi/nfs-pv/mount","volume_id":"nfs-server-nfs-server-provisioner.nfs-server.svc.cluste │
│ nfs I1219 15:22:14.849074       1 nodeserver.go:157] NodeUnpublishVolume: unmounting volume nfs-server-nfs-server-provisioner.nfs-server.svc.cluster.local/export/pvc-084cfb0d-cb0a-4089-a6ce-5b29049f4347 on /var/lib/kubelet/pods/bf38702d-92b6-40b │
│ nfs I1219 15:22:14.849084       1 nodeserver.go:162] force unmount nfs-server-nfs-server-provisioner.nfs-server.svc.cluster.local/export/pvc-084cfb0d-cb0a-4089-a6ce-5b29049f4347 on /var/lib/kubelet/pods/bf38702d-92b6-40b4-bfc0-1c7ab46f1633/volum │
│ nfs I1219 15:26:16.796186       1 utils.go:107] GRPC call: /csi.v1.Node/NodeUnpublishVolume                                                                                                                                                           │
│ nfs I1219 15:26:16.796327       1 utils.go:108] GRPC request: {"target_path":"/var/lib/kubelet/pods/4f08f426-6185-4e2a-9ced-83ea62e4d739/volumes/kubernetes.io~csi/nfs-pv/mount","volume_id":"nfs-server-nfs-server-provisioner.nfs-server.svc.cluste │
│ nfs I1219 15:26:16.796509       1 nodeserver.go:157] NodeUnpublishVolume: unmounting volume nfs-server-nfs-server-provisioner.nfs-server.svc.cluster.local/export/pvc-084cfb0d-cb0a-4089-a6ce-5b29049f4347 on /var/lib/kubelet/pods/4f08f426-6185-4e2 │
│ nfs I1219 15:26:16.796518       1 nodeserver.go:162] force unmount nfs-server-nfs-server-provisioner.nfs-server.svc.cluster.local/export/pvc-084cfb0d-cb0a-4089-a6ce-5b29049f4347 on /var/lib/kubelet/pods/4f08f426-6185-4e2a-9ced-83ea62e4d739/volum │
│ nfs I1219 15:26:16.896831       1 utils.go:107] GRPC call: /csi.v1.Node/NodeUnpublishVolume                                                                                                                                                           │
│ nfs I1219 15:26:16.896864       1 utils.go:108] GRPC request: {"target_path":"/var/lib/kubelet/pods/bf38702d-92b6-40b4-bfc0-1c7ab46f1633/volumes/kubernetes.io~csi/nfs-pv/mount","volume_id":"nfs-server-nfs-server-provisioner.nfs-server.svc.cluste │
│ nfs I1219 15:26:16.896926       1 nodeserver.go:157] NodeUnpublishVolume: unmounting volume nfs-server-nfs-server-provisioner.nfs-server.svc.cluster.local/export/pvc-084cfb0d-cb0a-4089-a6ce-5b29049f4347 on /var/lib/kubelet/pods/bf38702d-92b6-40b │
│ nfs I1219 15:26:16.896938       1 nodeserver.go:162] force unmount nfs-server-nfs-server-provisioner.nfs-server.svc.cluster.local/export/pvc-084cfb0d-cb0a-4089-a6ce-5b29049f4347 on /var/lib/kubelet/pods/bf38702d-92b6-40b4-bfc0-1c7ab46f1633/volum │
```

If a run a atop on the affected node it shows that the nfs-plugin keeps creating threads over time.
If i delete and recreate the csi-nfs-node pods the thread count goes back to normal.


**Environment**:
<!-- 
Run following command to get CSI driver version: 
kubectl get po -n kube-system -o yaml | grep gcr | grep nfs
-->
- CSI Driver version: [csi-driver-nfs-v4.5.0](image: registry.k8s.io/sig-storage/nfsplugin:v4.5.0)
- Kubernetes version (use `kubectl version`): Server Version: v1.28.2
- OS (e.g. from /etc/os-release): PRETTY_NAME="Debian GNU/Linux 12 (bookworm)"
- Kernel (e.g. `uname -a`): Linux tech-xrn0e 6.1.0-12-amd64 #1 SMP PREEMPT_DYNAMIC Debian 6.1.52-1 (2023-09-07) x86_64 GNU/Linux
- Install tools: flux/helm
- Others:


Pull request description:
fix: add locks for nodeserver publish/unpublish operations
<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our contributor guidelines: https://git.k8s.io/community/contributors/guide#your-first-contribution and developer guide https://git.k8s.io/community/contributors/devel/development.md#development-guide
2. Follow the instructions for writing a release note: https://git.k8s.io/community/contributors/guide/release-notes.md
3. If the PR is unfinished, see how to mark it: https://git.k8s.io/community/contributors/guide/pull-requests.md#marking-unfinished-pull-requests
-->

**What type of PR is this?**
/kind bug

**What this PR does / why we need it**:
fix: add locks for nodeserver publish/unpublish operations

Per CSI(Container Storage Interface), if a mount operation is in-progress for the volume then CSI driver should return error code "10 ABORTED"

**Which issue(s) this PR fixes**:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes #569 #209

**Special notes for your reviewer**:

**Does this PR introduce a user-facing change?**:
<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".
-->
```release-note
fix: add locks for nodeserver publish/unpublish operations
```


Patch:
diff --git a/pkg/nfs/nodeserver.go b/pkg/nfs/nodeserver.go
index 86559dbf..f9f5fc90 100644
--- a/pkg/nfs/nodeserver.go
+++ b/pkg/nfs/nodeserver.go
@@ -54,6 +54,13 @@ func (ns *NodeServer) NodePublishVolume(_ context.Context, req *csi.NodePublishV
 	if len(targetPath) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "Target path not provided")
 	}
+
+	lockKey := fmt.Sprintf("%s-%s", volumeID, targetPath)
+	if acquired := ns.Driver.volumeLocks.TryAcquire(lockKey); !acquired {
+		return nil, status.Errorf(codes.Aborted, volumeOperationAlreadyExistsFmt, volumeID)
+	}
+	defer ns.Driver.volumeLocks.Release(lockKey)
+
 	mountOptions := volCap.GetMount().GetMountFlags()
 	if req.GetReadonly() {
 		mountOptions = append(mountOptions, "ro")
@@ -156,6 +163,12 @@ func (ns *NodeServer) NodeUnpublishVolume(_ context.Context, req *csi.NodeUnpubl
 		return nil, status.Error(codes.InvalidArgument, "Target path missing in request")
 	}
 
+	lockKey := fmt.Sprintf("%s-%s", volumeID, targetPath)
+	if acquired := ns.Driver.volumeLocks.TryAcquire(lockKey); !acquired {
+		return nil, status.Errorf(codes.Aborted, volumeOperationAlreadyExistsFmt, volumeID)
+	}
+	defer ns.Driver.volumeLocks.Release(lockKey)
+
 	klog.V(2).Infof("NodeUnpublishVolume: unmounting volume %s on %s", volumeID, targetPath)
 	var err error
 	extensiveMountPointCheck := true
diff --git a/pkg/nfs/utils.go b/pkg/nfs/utils.go
index a8920fd7..f71605d2 100644
--- a/pkg/nfs/utils.go
+++ b/pkg/nfs/utils.go
@@ -34,10 +34,11 @@ import (
 
 //nolint:revive
 const (
-	separator = "#"
-	delete    = "delete"
-	retain    = "retain"
-	archive   = "archive"
+	separator                       = "#"
+	delete                          = "delete"
+	retain                          = "retain"
+	archive                         = "archive"
+	volumeOperationAlreadyExistsFmt = "An operation with the given Volume ID %s already exists"
 )
 
 var supportedOnDeleteValues = []string{"", delete, retain, archive}


Test patch:
diff --git a/pkg/nfs/nodeserver_test.go b/pkg/nfs/nodeserver_test.go
index dccb05f2..2d8bb9ca 100644
--- a/pkg/nfs/nodeserver_test.go
+++ b/pkg/nfs/nodeserver_test.go
@@ -19,6 +19,7 @@ package nfs
 import (
 	"context"
 	"errors"
+	"fmt"
 	"os"
 	"reflect"
 	"strings"
@@ -68,6 +69,7 @@ func TestNodePublishVolume(t *testing.T) {
 	volumeCap := csi.VolumeCapability_AccessMode{Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER}
 	alreadyMountedTarget := testutil.GetWorkDirPath("false_is_likely_exist_target", t)
 	targetTest := testutil.GetWorkDirPath("target_test", t)
+	lockKey := fmt.Sprintf("%s-%s", "vol_1", targetTest)
 
 	tests := []struct {
 		desc          string
@@ -93,6 +95,20 @@ func TestNodePublishVolume(t *testing.T) {
 				VolumeId: "vol_1"},
 			expectedErr: status.Error(codes.InvalidArgument, "Target path not provided"),
 		},
+		{
+			desc: "[Error] Volume operation in progress",
+			setup: func() {
+				ns.Driver.volumeLocks.TryAcquire(lockKey)
+			},
+			req: csi.NodePublishVolumeRequest{VolumeCapability: &csi.VolumeCapability{AccessMode: &volumeCap},
+				VolumeId:      "vol_1",
+				VolumeContext: params,
+				TargetPath:    targetTest},
+			expectedErr: status.Error(codes.Aborted, fmt.Sprintf(volumeOperationAlreadyExistsFmt, "vol_1")),
+			cleanup: func() {
+				ns.Driver.volumeLocks.Release(lockKey)
+			},
+		},
 		{
 			desc: "[Success] Stage target path missing",
 			req: csi.NodePublishVolumeRequest{
@@ -198,6 +214,7 @@ func TestNodeUnpublishVolume(t *testing.T) {
 	errorTarget := testutil.GetWorkDirPath("error_is_likely_target", t)
 	targetTest := testutil.GetWorkDirPath("target_test", t)
 	targetFile := testutil.GetWorkDirPath("abc.go", t)
+	lockKey := fmt.Sprintf("%s-%s", "vol_1", targetTest)
 
 	tests := []struct {
 		desc        string
@@ -220,6 +237,17 @@ func TestNodeUnpublishVolume(t *testing.T) {
 			desc: "[Success] Volume not mounted",
 			req:  csi.NodeUnpublishVolumeRequest{TargetPath: targetFile, VolumeId: "vol_1"},
 		},
+		{
+			desc: "[Error] Volume operation in progress",
+			setup: func() {
+				ns.Driver.volumeLocks.TryAcquire(lockKey)
+			},
+			req:         csi.NodeUnpublishVolumeRequest{TargetPath: targetTest, VolumeId: "vol_1"},
+			expectedErr: status.Error(codes.Aborted, fmt.Sprintf(volumeOperationAlreadyExistsFmt, "vol_1")),
+			cleanup: func() {
+				ns.Driver.volumeLocks.Release(lockKey)
+			},
+		},
 	}
 
 	// Setup