Repository: kubernetes-csi/csi-driver-nfs
Pull request #160 (2021-02-15 19:59:44)

feat: 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?**
> Uncomment only one ` /kind <>` line, hit enter to put that in a new line, and remove leading whitespaces from that line:
>
> /kind api-change
> /kind bug
> /kind cleanup
> /kind design
> /kind documentation
> /kind failing-test
> /kind feature
> /kind flake

/kind feature
**What this PR does / why we need it**:
Add locks for Node server publish/unpublish operations.
Only add locks for Node server because Controller server internally calls Node server.

**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 #

**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
none
```


k8s-ci-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-csi/csi-driver-nfs/pull/160#" title="Author self-approved">songjiaxun</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **saad-ali** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @saad-ali` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-csi%2Fcsi-driver-nfs).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/kubernetes-csi/csi-driver-nfs/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["saad-ali"]} -->
k8s-ci-robot: Hi @songjiaxun. Thanks for your PR.

I'm waiting for a [kubernetes-csi](https://github.com/orgs/kubernetes-csi/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://git.k8s.io/community/community-membership.md#member) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=kubernetes-csi%2Fcsi-driver-nfs).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>

songjiaxun: /assign @andyzhangx 
coveralls: ## Pull Request Test Coverage Report for [Build 569575526](https://coveralls.io/builds/37147666)

* **23** of **29**   **(79.31%)**  changed or added relevant lines in **3** files are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage decreased (**-0.01%**) to **79.589%**

---

|  Changes Missing Coverage | Covered Lines | Changed/Added Lines | % |
| :-----|--------------|--------|---: |
| [pkg/nfs/utils.go](https://coveralls.io/builds/37147666/source?filename=pkg%2Fnfs%2Futils.go#L113) | 16 | 18 | 88.89%
| [pkg/nfs/nodeserver.go](https://coveralls.io/builds/37147666/source?filename=pkg%2Fnfs%2Fnodeserver.go#L69) | 6 | 10 | 60.0%
<!-- | **Total:** | **23** | **29** | **79.31%** | -->


|  Totals | [![Coverage Status](https://coveralls.io/builds/37147666/badge)](https://coveralls.io/builds/37147666) |
| :-- | --: |
| Change from base [Build 556704704](https://coveralls.io/builds/37044767): |  -0.01% |
| Covered Lines: | 542 |
| Relevant Lines: | 681 |

---
##### 💛  - [Coveralls](https://coveralls.io)

andyzhangx: /ok-to-test
songjiaxun: > can you also port the unit test code? thanks.

Do you mean Node server unit test? I am porting Azure Disk, and I couldn't find any unit test related to the locks. 
andyzhangx: refer to https://github.com/kubernetes-sigs/azurefile-csi-driver/pull/513

Patch:
diff --git a/pkg/nfs/nfs.go b/pkg/nfs/nfs.go
index f6558c36..94adcce0 100644
--- a/pkg/nfs/nfs.go
+++ b/pkg/nfs/nfs.go
@@ -34,10 +34,11 @@ type Driver struct {
 	perm *uint32
 
 	//ids *identityServer
-	ns    *NodeServer
-	cap   map[csi.VolumeCapability_AccessMode_Mode]bool
-	cscap []*csi.ControllerServiceCapability
-	nscap []*csi.NodeServiceCapability
+	ns          *NodeServer
+	cap         map[csi.VolumeCapability_AccessMode_Mode]bool
+	cscap       []*csi.ControllerServiceCapability
+	nscap       []*csi.NodeServiceCapability
+	volumeLocks *VolumeLocks
 }
 
 const (
@@ -87,6 +88,7 @@ func NewNFSdriver(nodeID, endpoint string, perm *uint32) *Driver {
 		csi.NodeServiceCapability_RPC_GET_VOLUME_STATS,
 		csi.NodeServiceCapability_RPC_UNKNOWN,
 	})
+	n.volumeLocks = NewVolumeLocks()
 	return n
 }
 
diff --git a/pkg/nfs/nodeserver.go b/pkg/nfs/nodeserver.go
index cff6a48d..88c9ad84 100644
--- a/pkg/nfs/nodeserver.go
+++ b/pkg/nfs/nodeserver.go
@@ -65,6 +65,11 @@ func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis
 		return &csi.NodePublishVolumeResponse{}, nil
 	}
 
+	if acquired := ns.Driver.volumeLocks.TryAcquire(volumeID); !acquired {
+		return nil, status.Errorf(codes.Aborted, volumeOperationAlreadyExistsFmt, volumeID)
+	}
+	defer ns.Driver.volumeLocks.Release(volumeID)
+
 	mountOptions := req.GetVolumeCapability().GetMount().GetMountFlags()
 	if req.GetReadonly() {
 		mountOptions = append(mountOptions, "ro")
@@ -117,6 +122,11 @@ func (ns *NodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpu
 		return nil, status.Error(codes.NotFound, "Volume not mounted")
 	}
 
+	if acquired := ns.Driver.volumeLocks.TryAcquire(volumeID); !acquired {
+		return nil, status.Errorf(codes.Aborted, volumeOperationAlreadyExistsFmt, volumeID)
+	}
+	defer ns.Driver.volumeLocks.Release(volumeID)
+
 	klog.V(2).Infof("NodeUnpublishVolume: CleanupMountPoint %s on volumeID(%s)", targetPath, volumeID)
 	err = mount.CleanupMountPoint(targetPath, ns.mounter, false)
 	if err != nil {
diff --git a/pkg/nfs/utils.go b/pkg/nfs/utils.go
index c2d526ce..3b3c68ef 100644
--- a/pkg/nfs/utils.go
+++ b/pkg/nfs/utils.go
@@ -19,11 +19,13 @@ package nfs
 import (
 	"fmt"
 	"strings"
+	"sync"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
 	"github.com/kubernetes-csi/csi-lib-utils/protosanitizer"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
+	"k8s.io/apimachinery/pkg/util/sets"
 
 	"k8s.io/klog/v2"
 )
@@ -92,3 +94,34 @@ func logGRPC(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, h
 	}
 	return resp, err
 }
+
+const (
+	volumeOperationAlreadyExistsFmt = "An operation with the given Volume ID %s already exists"
+)
+
+type VolumeLocks struct {
+	locks sets.String
+	mux   sync.Mutex
+}
+
+func NewVolumeLocks() *VolumeLocks {
+	return &VolumeLocks{
+		locks: sets.NewString(),
+	}
+}
+
+func (vl *VolumeLocks) TryAcquire(volumeID string) bool {
+	vl.mux.Lock()
+	defer vl.mux.Unlock()
+	if vl.locks.Has(volumeID) {
+		return false
+	}
+	vl.locks.Insert(volumeID)
+	return true
+}
+
+func (vl *VolumeLocks) Release(volumeID string) {
+	vl.mux.Lock()
+	defer vl.mux.Unlock()
+	vl.locks.Delete(volumeID)
+}


Test patch:
diff --git a/pkg/nfs/nfs_test.go b/pkg/nfs/nfs_test.go
index db0d0338..e789a700 100644
--- a/pkg/nfs/nfs_test.go
+++ b/pkg/nfs/nfs_test.go
@@ -59,7 +59,7 @@ func NewEmptyDriver(emptyField string) *Driver {
 			perm:    perm,
 		}
 	}
-
+	d.volumeLocks = NewVolumeLocks()
 	return d
 }
 
diff --git a/pkg/nfs/nodeserver_test.go b/pkg/nfs/nodeserver_test.go
index 6a0cb633..5bd2a01d 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"
 	"testing"
@@ -35,15 +36,22 @@ const (
 )
 
 func TestNodePublishVolume(t *testing.T) {
+	ns, err := getTestNodeServer()
+	if err != nil {
+		t.Fatalf(err.Error())
+	}
+
 	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)
 
 	tests := []struct {
 		desc          string
+		setup         func()
 		req           csi.NodePublishVolumeRequest
 		skipOnWindows bool
 		expectedErr   error
+		cleanup       func()
 	}{
 		{
 			desc:        "[Error] Volume capabilities missing",
@@ -61,6 +69,19 @@ 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("vol_1")
+			},
+			req: csi.NodePublishVolumeRequest{VolumeCapability: &csi.VolumeCapability{AccessMode: &volumeCap},
+				VolumeId:   "vol_1",
+				TargetPath: targetTest},
+			expectedErr: status.Error(codes.Aborted, fmt.Sprintf(volumeOperationAlreadyExistsFmt, "vol_1")),
+			cleanup: func() {
+				ns.Driver.volumeLocks.Release("vol_1")
+			},
+		},
 		{
 			desc: "[Success] Stage target path missing",
 			req: csi.NodePublishVolumeRequest{VolumeCapability: &csi.VolumeCapability{AccessMode: &volumeCap},
@@ -97,16 +118,17 @@ func TestNodePublishVolume(t *testing.T) {
 	// setup
 	_ = makeDir(alreadyMountedTarget)
 
-	ns, err := getTestNodeServer()
-	if err != nil {
-		t.Fatalf(err.Error())
-	}
-
 	for _, tc := range tests {
+		if tc.setup != nil {
+			tc.setup()
+		}
 		_, err := ns.NodePublishVolume(context.Background(), &tc.req)
 		if !reflect.DeepEqual(err, tc.expectedErr) {
 			t.Errorf("Desc:%v\nUnexpected error: %v\nExpected: %v", tc.desc, err, tc.expectedErr)
 		}
+		if tc.cleanup != nil {
+			tc.cleanup()
+		}
 	}
 
 	// Clean up
@@ -118,14 +140,22 @@ func TestNodePublishVolume(t *testing.T) {
 }
 
 func TestNodeUnpublishVolume(t *testing.T) {
+	ns, err := getTestNodeServer()
+	if err != nil {
+		t.Fatalf(err.Error())
+	}
+
 	errorTarget := testutil.GetWorkDirPath("error_is_likely_target", t)
 	targetTest := testutil.GetWorkDirPath("target_test", t)
 	targetFile := testutil.GetWorkDirPath("abc.go", t)
+	alreadyMountedTarget := testutil.GetWorkDirPath("false_is_likely_exist_target", t)
 
 	tests := []struct {
 		desc        string
+		setup       func()
 		req         csi.NodeUnpublishVolumeRequest
 		expectedErr error
+		cleanup     func()
 	}{
 		{
 			desc:        "[Error] Volume ID missing",
@@ -147,21 +177,33 @@ func TestNodeUnpublishVolume(t *testing.T) {
 			req:         csi.NodeUnpublishVolumeRequest{TargetPath: targetFile, VolumeId: "vol_1"},
 			expectedErr: status.Error(codes.NotFound, "Volume not mounted"),
 		},
+		{
+			desc: "[Error] Volume operation in progress",
+			setup: func() {
+				ns.Driver.volumeLocks.TryAcquire("vol_1")
+			},
+			req:         csi.NodeUnpublishVolumeRequest{TargetPath: alreadyMountedTarget, VolumeId: "vol_1"},
+			expectedErr: status.Error(codes.Aborted, fmt.Sprintf(volumeOperationAlreadyExistsFmt, "vol_1")),
+			cleanup: func() {
+				ns.Driver.volumeLocks.Release("vol_1")
+			},
+		},
 	}
 
 	// Setup
 	_ = makeDir(errorTarget)
 
-	ns, err := getTestNodeServer()
-	if err != nil {
-		t.Fatalf(err.Error())
-	}
-
 	for _, tc := range tests {
+		if tc.setup != nil {
+			tc.setup()
+		}
 		_, err := ns.NodeUnpublishVolume(context.Background(), &tc.req)
 		if !reflect.DeepEqual(err, tc.expectedErr) {
 			t.Errorf("Desc:%v\nUnexpected error: %v\nExpected: %v", tc.desc, err, tc.expectedErr)
 		}
+		if tc.cleanup != nil {
+			tc.cleanup()
+		}
 	}
 
 	// Clean up