Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: add locks for nodeserver publish/unpublish operations #160

Merged
merged 1 commit into from
Feb 17, 2021
Merged

feat: add locks for nodeserver publish/unpublish operations #160

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 6 additions & 4 deletions pkg/nfs/nfs.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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
}

Expand Down
2 changes: 1 addition & 1 deletion pkg/nfs/nfs_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ func NewEmptyDriver(emptyField string) *Driver {
perm: perm,
}
}

d.volumeLocks = NewVolumeLocks()
return d
}

Expand Down
10 changes: 10 additions & 0 deletions pkg/nfs/nodeserver.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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 {
Expand Down
62 changes: 52 additions & 10 deletions pkg/nfs/nodeserver_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ package nfs
import (
"context"
"errors"
"fmt"
"os"
"reflect"
"testing"
Expand All @@ -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",
Expand All @@ -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},
Expand Down Expand Up @@ -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
Expand All @@ -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",
Expand All @@ -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
Expand Down
33 changes: 33 additions & 0 deletions pkg/nfs/utils.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)
Expand Down Expand Up @@ -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)
}