add mutex

go.mod

@@ -20,6 +20,7 @@ require (
 	google.golang.org/grpc v1.23.0
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	k8s.io/apiextensions-apiserver v0.0.0-20190823014223-07b4561f8b0e // indirect
+	k8s.io/apimachinery v0.0.0-20190823012420-8ca64af22337
 	k8s.io/apiserver v0.0.0-20190823053033-1316076af51c // indirect
 	k8s.io/cloud-provider v0.0.0-20190717025205-585d8110a88f // indirect
 	k8s.io/klog v0.4.0

pkg/common/lock.go

@@ -0,0 +1,57 @@
+/*
+Copyright (C) 2018 Yunify, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this work except in compliance with the License.
+You may obtain a copy of the License in the LICENSE file, or at:
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package common
+
+import (
+	"k8s.io/apimachinery/pkg/util/sets"
+	"sync"
+)
+
+const (
+	OperationPendingFmt = "already an operation for the specified resource %s"
+)
+
+// ResourceLocks implements a map with atomic operations. It stores a set of all resource IDs
+// with an ongoing operation.
+type ResourceLocks struct {
+	locks sets.String
+	mux   sync.Mutex
+}
+
+func NewResourceLocks() *ResourceLocks {
+	return &ResourceLocks{
+		locks: sets.NewString(),
+	}
+}
+
+// TryAcquire tries to acquire the lock for operating on resourceID and returns true if successful.
+// If another operation is already using resourceID, returns false.
+func (lock *ResourceLocks) TryAcquire(resourceID string) bool {
+	lock.mux.Lock()
+	defer lock.mux.Unlock()
+	if lock.locks.Has(resourceID) {
+		return false
+	}
+	lock.locks.Insert(resourceID)
+	return true
+}
+
+func (lock *ResourceLocks) Release(resourceID string) {
+	lock.mux.Lock()
+	defer lock.mux.Unlock()
+	lock.locks.Delete(resourceID)
+}

pkg/common/lock_test.go

@@ -0,0 +1,49 @@
+package common
+
+import "testing"
+
+func TestResourceLocks_TryAcquire(t *testing.T) {
+	lock := NewResourceLocks()
+	tests := []struct {
+		name       string
+		resourceId string
+		isAcquire  bool
+	}{
+		{
+			name:       "first acquire",
+			resourceId: "vol-123456",
+			isAcquire:  true,
+		},
+		{
+			name:       "acquire failed",
+			resourceId: "vol-123456",
+			isAcquire:  false,
+		},
+		{
+			name:       "acquire another",
+			resourceId: "vol-234567",
+			isAcquire:  true,
+		},
+	}
+	for _, test := range tests {
+		res := lock.TryAcquire(test.resourceId)
+		if test.isAcquire != res {
+			t.Errorf("name %s: expect %t, but actually %t", test.name, test.isAcquire, res)
+		}
+	}
+}
+
+func TestResourceLocks_Release(t *testing.T) {
+	lock := NewResourceLocks()
+	resourceId1 := "vol-00001"
+	resourceId2 := "vol-00002"
+	// release a not exist resource
+	lock.Release(resourceId1)
+	// release a exist resource
+	if res := lock.TryAcquire(resourceId2); !res {
+		t.Errorf("try acquire %s failed", resourceId2)
+	}
+	lock.Release(resourceId2)
+	// release multiple times
+	lock.Release(resourceId2)
+}

pkg/disk/rpcserver/controllerserver.go

@@ -37,7 +37,7 @@ import (
 type ControllerServer struct {
 	driver *driver.DiskDriver
 	cloud  cloud.CloudManager
-	// TODO: add mutex
+	locks  *common.ResourceLocks
 }
 
 // NewControllerServer
@@ -46,9 +46,12 @@ func NewControllerServer(d *driver.DiskDriver, c cloud.CloudManager) *Controller
 	return &ControllerServer{
 		driver: d,
 		cloud:  c,
+		locks:  common.NewResourceLocks(),
 	}
 }
 
+var _ csi.ControllerServer = &ControllerServer{}
+
 // This operation MUST be idempotent
 // This operation MAY create three types of volumes:
 // 1. Empty volumes: CREATE_DELETE_VOLUME
@@ -282,6 +285,13 @@ func (cs *ControllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVol
 	// Deleting disk
 	klog.Infof("deleting volume %s", volumeId)
 
+	// ensure on call in-flight
+	klog.Infof("try to lock resource %s", volumeId)
+	if acquired := cs.locks.TryAcquire(volumeId); !acquired {
+		return nil, status.Errorf(codes.Aborted, common.OperationPendingFmt, volumeId)
+	}
+	defer cs.locks.Release(volumeId)
+
 	// For idempotent:
 	// MUST reply OK when volume does not exist
 	volInfo, err := cs.cloud.FindVolume(volumeId)
@@ -345,6 +355,14 @@ func (cs *ControllerServer) ControllerPublishVolume(ctx context.Context, req *cs
 
 	// if volume id not exist
 	volumeId := req.GetVolumeId()
+
+	// ensure one call in-flight
+	klog.Infof("try to lock resource %s", volumeId)
+	if acquired := cs.locks.TryAcquire(volumeId); !acquired {
+		return nil, status.Errorf(codes.Aborted, common.OperationPendingFmt, volumeId)
+	}
+	defer cs.locks.Release(volumeId)
+
 	exVol, err := cs.cloud.FindVolume(volumeId)
 	if err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
@@ -442,6 +460,13 @@ func (cs *ControllerServer) ControllerUnpublishVolume(ctx context.Context, req *
 	volumeId := req.GetVolumeId()
 	nodeId := req.GetNodeId()
 
+	// ensure one call in-flight
+	klog.Infof("try to lock resource %s", volumeId)
+	if acquired := cs.locks.TryAcquire(volumeId); !acquired {
+		return nil, status.Errorf(codes.Aborted, common.OperationPendingFmt, volumeId)
+	}
+	defer cs.locks.Release(volumeId)
+
 	// 1. Detach
 	// check volume exist
 	exVol, err := cs.cloud.FindVolume(volumeId)
@@ -539,6 +564,13 @@ func (cs *ControllerServer) ControllerExpandVolume(ctx context.Context, req *csi
 	// 1. Check volume status
 	// does volume exist
 	volumeId := req.GetVolumeId()
+	// ensure one call in-flight
+	klog.Infof("try to lock resource %s", volumeId)
+	if acquired := cs.locks.TryAcquire(volumeId); !acquired {
+		return nil, status.Errorf(codes.Aborted, common.OperationPendingFmt, volumeId)
+	}
+	defer cs.locks.Release(volumeId)
+
 	volInfo, err := cs.cloud.FindVolume(volumeId)
 	if err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
@@ -625,6 +657,12 @@ func (cs *ControllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateS
 	// Create snapshot manager object
 	srcVolId := req.GetSourceVolumeId()
 	snapName := req.GetName()
+	// ensure one call in-flight
+	klog.Infof("try to lock resource %s", srcVolId)
+	if acquired := cs.locks.TryAcquire(srcVolId); !acquired {
+		return nil, status.Errorf(codes.Aborted, common.OperationPendingFmt, srcVolId)
+	}
+	defer cs.locks.Release(srcVolId)
 	var ts *timestamp.Timestamp
 	var isReadyToUse bool
 	// For idempotent
@@ -738,6 +776,12 @@ func (cs *ControllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteS
 		return nil, status.Error(codes.InvalidArgument, "snapshot ID missing in request")
 	}
 	snapId := req.GetSnapshotId()
+	// ensure one call in-flight
+	klog.Infof("try to lock resource %s", snapId)
+	if acquired := cs.locks.TryAcquire(snapId); !acquired {
+		return nil, status.Errorf(codes.Aborted, common.OperationPendingFmt, snapId)
+	}
+	defer cs.locks.Release(snapId)
 	// 1. For idempotent:
 	// MUST reply OK when snapshot does not exist
 	klog.Infof("Find existing snapshot id [%s].", snapId)

pkg/disk/rpcserver/identityserver.go

@@ -26,22 +26,24 @@ import (
 	"k8s.io/klog"
 )
 
-type DiskIdentityServer struct {
+type IdentityServer struct {
 	driver *driver.DiskDriver
 	cloud  cloud.CloudManager
 }
 
 // NewIdentityServer
 // Create identity server
-func NewIdentityServer(d *driver.DiskDriver, c cloud.CloudManager) *DiskIdentityServer {
-	return &DiskIdentityServer{
+func NewIdentityServer(d *driver.DiskDriver, c cloud.CloudManager) *IdentityServer {
+	return &IdentityServer{
 		driver: d,
 		cloud:  c,
 	}
 }
 
+var _ csi.IdentityServer = &IdentityServer{}
+
 // Plugin MUST implement this RPC call
-func (is *DiskIdentityServer) Probe(ctx context.Context, req *csi.ProbeRequest) (*csi.ProbeResponse, error) {
+func (is *IdentityServer) Probe(ctx context.Context, req *csi.ProbeRequest) (*csi.ProbeResponse, error) {
 	zones, err := is.cloud.GetZoneList()
 	if err != nil {
 		return nil, status.Error(codes.FailedPrecondition, err.Error())
@@ -51,15 +53,15 @@ func (is *DiskIdentityServer) Probe(ctx context.Context, req *csi.ProbeRequest)
 }
 
 // Get plugin capabilities: CONTROLLER, ACCESSIBILITY, EXPANSION
-func (d *DiskIdentityServer) GetPluginCapabilities(ctx context.Context, req *csi.GetPluginCapabilitiesRequest) (*csi.
+func (d *IdentityServer) GetPluginCapabilities(ctx context.Context, req *csi.GetPluginCapabilitiesRequest) (*csi.
 	GetPluginCapabilitiesResponse, error) {
 	klog.V(5).Infof("Using default capabilities")
 	return &csi.GetPluginCapabilitiesResponse{
 		Capabilities: d.driver.GetPluginCapability(),
 	}, nil
 }
 
-func (d *DiskIdentityServer) GetPluginInfo(ctx context.Context,
+func (d *IdentityServer) GetPluginInfo(ctx context.Context,
 	req *csi.GetPluginInfoRequest) (*csi.GetPluginInfoResponse, error) {
 	klog.V(5).Infof("Using GetPluginInfo")