Repository: topolvm/topolvm
Pull request #659 (2023-02-17 13:50:09)

Clarify lock usage
We use some mutexes, but it is unclear what for. This PR adds comments or refactors code to improve meanings.
The most notable part is moving a locking functionality from LogicalVolumeService to Node/ControllerServer. Also, note that the new lock type `LockByID` improves parallelism.


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index edbf20a0..a644ab57 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,38 +7,8 @@ This file itself is based on [Keep a CHANGELOG](https://keepachangelog.com/en/0.
 
 ## [Unreleased]
 
-## [0.18.0] - 2023-02-20
-
-### Added
-
-- add proposal to specify lvcreate options on SC ([#627](https://github.com/topolvm/topolvm/pull/627))
-- Add CONTRIBUTING.md ([#631](https://github.com/topolvm/topolvm/pull/631))
-- add a note describing how to maintain go version ([#633](https://github.com/topolvm/topolvm/pull/633))
-- Add the lvcreate-option-on-storageclass proposal implementation ([#640](https://github.com/topolvm/topolvm/pull/640))
-- artifacthub ([#641](https://github.com/topolvm/topolvm/pull/641))
-
-### Changed
-
-- Revert "Drop a PVC finalizer to delete pods(#536)" ([#620](https://github.com/topolvm/topolvm/pull/620))
+- Revert "add a test to check the PVC finalizer is deleted." ([#620](https://github.com/topolvm/topolvm/pull/620))
   - **Note**: The PVC finalizer is not added to the existing PVCs. If the problem explained in [issue #614](https://github.com/topolvm/topolvm/issues/614) happens for those PVCs, you need to resolve it manually.
-- build(deps): bump actions/stale from 6 to 7 ([#628](https://github.com/topolvm/topolvm/pull/628))
-- Update go directive and use the version for setup-go ([#629](https://github.com/topolvm/topolvm/pull/629))
-- Make CI faster again ([#638](https://github.com/topolvm/topolvm/pull/638))
-- try to update go 1.19 to fix ci ([#652](https://github.com/topolvm/topolvm/pull/652))
-
-### Fixed
-
-- fix the proposal ([#642](https://github.com/topolvm/topolvm/pull/642))
-- fix: set default capacity to thin class capacity when default ([#632](https://github.com/topolvm/topolvm/pull/632))
-
-### Contributors
-
-- @bells17
-- @cupnes
-- @llamerada-jp
-- @peng225
-- @suleymanakbas91
-- @toshipp
 
 ## [0.17.0] - 2023-01-10
 
@@ -705,8 +675,7 @@ Only cosmetic changes.
 
 This is the first release.
 
-[Unreleased]: https://github.com/topolvm/topolvm/compare/v0.18.0...HEAD
-[0.18.0]: https://github.com/topolvm/topolvm/compare/v0.17.0...v0.18.0
+[Unreleased]: https://github.com/topolvm/topolvm/compare/v0.17.0...HEAD
 [0.17.0]: https://github.com/topolvm/topolvm/compare/v0.16.0...v0.17.0
 [0.16.0]: https://github.com/topolvm/topolvm/compare/v0.15.3...v0.16.0
 [0.15.3]: https://github.com/topolvm/topolvm/compare/v0.15.2...v0.15.3
diff --git a/Makefile b/Makefile
index d1bf7e9f..67c7de0a 100644
--- a/Makefile
+++ b/Makefile
@@ -113,7 +113,7 @@ manifests: generate-legacy-api ## Generate WebhookConfiguration, ClusterRole and
 		crd:crdVersions=v1 \
 		rbac:roleName=topolvm-controller \
 		webhook \
-		paths="./api/...;./controllers;./hook;./driver/k8s;./pkg/..." \
+		paths="./api/...;./controllers;./hook;./driver/internal/k8s;./pkg/..." \
 		output:crd:artifacts:config=config/crd/bases
 	$(BINDIR)/yq eval 'del(.status)' config/crd/bases/topolvm.io_logicalvolumes.yaml | xargs -d"	" printf "$$CRD_TEMPLATE" > charts/topolvm/templates/crds/topolvm.io_logicalvolumes.yaml
 	$(BINDIR)/yq eval 'del(.status)' config/crd/bases/topolvm.cybozu.com_logicalvolumes.yaml | xargs -d"	" printf "$$LEGACY_CRD_TEMPLATE" > charts/topolvm/templates/crds/topolvm.cybozu.com_logicalvolumes.yaml
diff --git a/RELEASE.md b/RELEASE.md
index ef4b13a0..d0550ee3 100644
--- a/RELEASE.md
+++ b/RELEASE.md
@@ -55,22 +55,29 @@ Bump version
     $ git checkout -b bump-$VERSION
     ```
 
-3. Edit `CHANGELOG.md` for the new version ([example][]).
+3. Update image versions in files below:
+   - example/Makefile
+   - example/README.md
+    ```console
+    $ sed -r -i "s/TOPOLVM_VERSION := [[:digit:]]+\.[[:digit:]]+\.[[:digit:]]+/TOPOLVM_VERSION := ${VERSION}/g" example/Makefile
+    $ sed -r -i "s/checkout v[[:digit:]]+\.[[:digit:]]+\.[[:digit:]]+/checkout v${VERSION}/g" example/README.md
+    ```
+4. Edit `CHANGELOG.md` for the new version ([example][]).
    The candidate of relevant PRs can be obtained by the following command.
    ```
    git log --merges --format="%s%x00%b" $(git tag | grep "^v" | sort -V -r | head -n 1)..main | sed -E 's|^Merge pull request #([0-9]*)[^\x0]*\x0(.*)|- \2 ([#\1](https://github.com/topolvm/topolvm/pull/\1))|' | tac
    ```
    Please remove PRs which contain changes only to the helm chart.
 
-4. Commit the change and create a pull request:
+5. Commit the change and create a pull request:
 
     ```console
     $ git commit -a -s -m "Bump version to $VERSION"
     $ git push -u origin bump-$VERSION
     ```
 
-5. Create new pull request and merge it.
-6. Add a new tag and push it as follows:
+6. Create new pull request and merge it.
+7. Add a new tag and push it as follows:
 
     ```console
     $ git checkout main
@@ -113,12 +120,8 @@ This will prevent the TopoLVM version from going up just by modifying the Helm C
     ```
 
 3. Update image versions in files below:
-   - example/Makefile
-   - example/README.md
    - charts/topolvm/Chart.yaml
     ```console
-    $ sed -r -i "s/TOPOLVM_VERSION := [[:digit:]]+\.[[:digit:]]+\.[[:digit:]]+/TOPOLVM_VERSION := ${APPVERSION}/g" example/Makefile
-    $ sed -r -i "s/checkout v[[:digit:]]+\.[[:digit:]]+\.[[:digit:]]+/checkout v${APPVERSION}/g" example/README.md
     $ sed -r -i "s/appVersion: [[:digit:]]+\.[[:digit:]]+\.[[:digit:]]+/appVersion: ${APPVERSION}/g" charts/topolvm/Chart.yaml
     $ sed -r -i "s/^version: [[:digit:]]+\.[[:digit:]]+\.[[:digit:]]+/version: ${CHARTVERSION}/g" charts/topolvm/Chart.yaml
     $ sed -r -i "s/ghcr.io\/topolvm\/topolvm-with-sidecar:[[:digit:]]+\.[[:digit:]]+\.[[:digit:]]+/ghcr.io\/topolvm\/topolvm-with-sidecar:${APPVERSION}/g" charts/topolvm/Chart.yaml
diff --git a/charts/topolvm/CHANGELOG.md b/charts/topolvm/CHANGELOG.md
index e1fe7e5c..73672945 100644
--- a/charts/topolvm/CHANGELOG.md
+++ b/charts/topolvm/CHANGELOG.md
@@ -7,30 +7,6 @@ This file itself is based on [Keep a CHANGELOG](https://keepachangelog.com/en/0.
 
 ## [Unreleased]
 
-## [11.1.0] - 2023-02-20
-
-### Added
-
-- Revert "Drop a PVC finalizer to delete pods(#536)" ([#620](https://github.com/topolvm/topolvm/pull/620))
-- Add the lvcreate-option-on-storageclass proposal implementation ([#640](https://github.com/topolvm/topolvm/pull/640))
-- artifacthub ([#641](https://github.com/topolvm/topolvm/pull/641))
-
-### Changed
-
-- Remove metrics annotation from lvmd daemonset ([#643](https://github.com/topolvm/topolvm/pull/643))
-- build(deps): bump helm/chart-releaser-action from 1.4.1 to 1.5.0 ([#649](https://github.com/topolvm/topolvm/pull/649))
-
-### Fixed
-
-- fix($charts):snapshotter ClusterRoleBinding name error #635 ([#637](https://github.com/topolvm/topolvm/pull/637))
-
-### Contributors
-
-- @bells17
-- @peng225
-- @syang1997
-- @Thor77
-
 ## [11.0.1] - 2022-01-10
 
 ### Changed
@@ -260,8 +236,7 @@ This release contains the domain name change([#592](https://github.com/topolvm/t
 
 This is the first release.
 
-[Unreleased]: https://github.com/topolvm/topolvm/compare/topolvm-chart-v11.1.0...HEAD
-[11.1.0]: https://github.com/topolvm/topolvm/compare/topolvm-chart-v11.0.1...topolvm-chart-v11.1.0
+[Unreleased]: https://github.com/topolvm/topolvm/compare/topolvm-chart-v11.0.1...HEAD
 [11.0.1]: https://github.com/topolvm/topolvm/compare/topolvm-chart-v11.0.0...topolvm-chart-v11.0.1
 [11.0.0]: https://github.com/topolvm/topolvm/compare/topolvm-chart-v10.0.0...topolvm-chart-v11.0.0
 [10.0.0]: https://github.com/topolvm/topolvm/compare/topolvm-chart-v9.1.0...topolvm-chart-v10.0.0
diff --git a/charts/topolvm/Chart.yaml b/charts/topolvm/Chart.yaml
index 411002cd..c9c7032a 100644
--- a/charts/topolvm/Chart.yaml
+++ b/charts/topolvm/Chart.yaml
@@ -4,8 +4,8 @@ type: application
 home: https://github.com/topolvm/topolvm
 name: topolvm
 description: Topolvm
-version: 11.1.0
-appVersion: 0.18.0
+version: 11.0.1
+appVersion: 0.17.0
 sources:
   - https://github.com/topolvm/topolvm
 
@@ -18,7 +18,7 @@ dependencies:
 annotations:
   artifacthub.io/images: |
     - name: topolvm-with-sidecar
-      image: ghcr.io/topolvm/topolvm-with-sidecar:0.18.0
+      image: ghcr.io/topolvm/topolvm-with-sidecar:0.17.0
     - name: topolvm-with
-      image: ghcr.io/topolvm/topolvm:0.18.0
+      image: ghcr.io/topolvm/topolvm:0.17.0
   artifacthub.io/license: Apache-2.0
diff --git a/csi-sidecars.mk b/csi-sidecars.mk
index 1a459433..51c6e61b 100644
--- a/csi-sidecars.mk
+++ b/csi-sidecars.mk
@@ -20,7 +20,6 @@ EXTERNAL_RESIZER_SRC      = $(SRC_ROOT)/external-resizer
 EXTERNAL_SNAPSHOTTER_SRC  = $(SRC_ROOT)/external-snapshotter
 LIVENESSPROBE_SRC         = $(SRC_ROOT)/livenessprobe
 
-MAKEFILE_DIR := $(dir $(firstword $(MAKEFILE_LIST)))
 OUTPUT_DIR ?= .
 
 CURL := curl -sSLf
@@ -35,7 +34,7 @@ $(OUTPUT_DIR)/.csi-provisioner-$(EXTERNAL_PROVISIONER_VERSION):
 	mkdir -p $(EXTERNAL_PROVISIONER_SRC)
 	$(CURL) https://github.com/kubernetes-csi/external-provisioner/archive/v$(EXTERNAL_PROVISIONER_VERSION).tar.gz | \
         tar zxf - --strip-components 1 -C $(EXTERNAL_PROVISIONER_SRC)
-	patch -d $(EXTERNAL_PROVISIONER_SRC)/release-tools < $(MAKEFILE_DIR)/cache-packages.patch
+	patch -d $(EXTERNAL_PROVISIONER_SRC)/release-tools < cache-packages.patch
 	make -C $(EXTERNAL_PROVISIONER_SRC)
 	cp -f $(EXTERNAL_PROVISIONER_SRC)/bin/csi-provisioner $@
 
@@ -47,7 +46,7 @@ $(OUTPUT_DIR)/.csi-snapshotter-$(EXTERNAL_SNAPSHOTTER_VERSION):
 	mkdir -p $(EXTERNAL_SNAPSHOTTER_SRC)
 	curl -sSLf https://github.com/kubernetes-csi/external-snapshotter/archive/v$(EXTERNAL_SNAPSHOTTER_VERSION).tar.gz | \
         tar zxf - --strip-components 1 -C $(EXTERNAL_SNAPSHOTTER_SRC)
-	patch -d $(EXTERNAL_SNAPSHOTTER_SRC)/release-tools < $(MAKEFILE_DIR)/cache-packages.patch
+	patch -d $(EXTERNAL_SNAPSHOTTER_SRC)/release-tools < cache-packages.patch
 	make -C $(EXTERNAL_SNAPSHOTTER_SRC)
 	cp -f $(EXTERNAL_SNAPSHOTTER_SRC)/bin/csi-snapshotter $@
 
@@ -59,7 +58,7 @@ $(OUTPUT_DIR)/.csi-resizer-$(EXTERNAL_RESIZER_VERSION):
 	mkdir -p $(EXTERNAL_RESIZER_SRC)
 	$(CURL) https://github.com/kubernetes-csi/external-resizer/archive/v$(EXTERNAL_RESIZER_VERSION).tar.gz | \
         tar zxf - --strip-components 1 -C $(EXTERNAL_RESIZER_SRC)
-	patch -d $(EXTERNAL_RESIZER_SRC)/release-tools < $(MAKEFILE_DIR)/cache-packages.patch
+	patch -d $(EXTERNAL_RESIZER_SRC)/release-tools < cache-packages.patch
 	make -C $(EXTERNAL_RESIZER_SRC)
 	cp -f $(EXTERNAL_RESIZER_SRC)/bin/csi-resizer $@
 
@@ -71,7 +70,7 @@ $(OUTPUT_DIR)/.csi-node-driver-registrar-$(NODE_DRIVER_REGISTRAR_VERSION):
 	mkdir -p $(NODE_DRIVER_REGISTRAR_SRC)
 	$(CURL) https://github.com/kubernetes-csi/node-driver-registrar/archive/v$(NODE_DRIVER_REGISTRAR_VERSION).tar.gz | \
         tar zxf - --strip-components 1 -C $(NODE_DRIVER_REGISTRAR_SRC)
-	patch -d $(NODE_DRIVER_REGISTRAR_SRC)/release-tools < $(MAKEFILE_DIR)/cache-packages.patch
+	patch -d $(NODE_DRIVER_REGISTRAR_SRC)/release-tools < cache-packages.patch
 	make -C $(NODE_DRIVER_REGISTRAR_SRC)
 	cp -f $(NODE_DRIVER_REGISTRAR_SRC)/bin/csi-node-driver-registrar $@
 
@@ -83,7 +82,7 @@ $(OUTPUT_DIR)/.livenessprobe-$(LIVENESSPROBE_VERSION):
 	mkdir -p $(LIVENESSPROBE_SRC)
 	$(CURL) https://github.com/kubernetes-csi/livenessprobe/archive/v$(LIVENESSPROBE_VERSION).tar.gz | \
         tar zxf - --strip-components 1 -C $(LIVENESSPROBE_SRC)
-	patch -d $(LIVENESSPROBE_SRC)/release-tools < $(MAKEFILE_DIR)/cache-packages.patch
+	patch -d $(LIVENESSPROBE_SRC)/release-tools < cache-packages.patch
 	make -C $(LIVENESSPROBE_SRC)
 	cp -f $(LIVENESSPROBE_SRC)/bin/livenessprobe $@
 
diff --git a/docs/cleanup.md b/docs/cleanup.md
deleted file mode 100644
index f003f578..00000000
--- a/docs/cleanup.md
+++ /dev/null
@@ -1,35 +0,0 @@
-# Cleanup
-
-To cleanup TopoLVM, please follow these steps.
-
-**Caution: The cleanup procedure cannot be reverted. Think twice whether you can really cleanup TopoLVM.**
-
-1. Delete all the snapshots and PVCs related to TopoLVM.
-2. If the scheduler extender is enabled, edit the kube-scheduler's config file and remove TopoLVM's settings. It may be your help to read [the configuration guide](https://github.com/topolvm/topolvm/blob/main/deploy/README.md#configure-kube-scheduler).
-   - Please be careful if you have any other settings in extenders section of the configuration file. You can remove only the TopoLVM's setting.
-3. Uninstall TopoLVM via Helm.
-   ```console
-   # Check the namespace
-   helm list -A
-   
-   # Uninstall. Here we assume the namespace is `topolvm-system`.
-   helm uninstall --namespace=topolvm-system topolvm
-   ```
-4. Remove a label from the target namespace.  
-   - When legacy mode is disabled:
-     ```console
-     # Please change the namespace depending on your environment.
-     kubectl label namespace topolvm-system topolvm.io/webhook-
-     kubectl label namespace kube-system topolvm.io/webhook-
-     ```
-   - When legacy mode is enabled:
-     ```console
-     # Please change the namespace depending on your environment.
-     kubectl label namespace topolvm-system topolvm.cybozu.com/webhook-
-     kubectl label namespace kube-system topolvm.cybozu.com/webhook-
-     ```
-5. If lvmd is running as a systemd unit, stop it on each node.
-   ```console
-   systemctl is-active lvmd.service
-   systemctl disable --now lvmd.service
-   ```
diff --git a/driver/controller.go b/driver/controller.go
index 7a00c321..fa977885 100644
--- a/driver/controller.go
+++ b/driver/controller.go
@@ -11,27 +11,107 @@ import (
 	"github.com/topolvm/topolvm"
 	v1 "github.com/topolvm/topolvm/api/v1"
 	"github.com/topolvm/topolvm/csi"
-	"github.com/topolvm/topolvm/driver/k8s"
+	"github.com/topolvm/topolvm/driver/internal/k8s"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 )
 
 var ctrlLogger = ctrl.Log.WithName("driver").WithName("controller")
 
-// NewControllerService returns a new ControllerServer.
-func NewControllerService(lvService *k8s.LogicalVolumeService, nodeService *k8s.NodeService) csi.ControllerServer {
-	return &controllerService{lvService: lvService, nodeService: nodeService}
+// NewControllerServer returns a new ControllerServer.
+func NewControllerServer(mgr manager.Manager) (csi.ControllerServer, error) {
+	lvService, err := k8s.NewLogicalVolumeService(mgr)
+	if err != nil {
+		return nil, err
+	}
+
+	return &controllerServer{
+		lockByName:     NewLockWithID(),
+		lockByVolumeID: NewLockWithID(),
+		server: &controllerServerNoLocked{
+			lvService:   lvService,
+			nodeService: k8s.NewNodeService(mgr.GetClient()),
+		},
+	}, nil
+}
+
+// This is a wrapper for controllerServerNoLocked to protect concurrent method call.
+type controllerServer struct {
+	csi.UnimplementedControllerServer
+
+	// This protects server methods using a volume name.
+	lockByName *LockByID
+	// This protects server methods using a volume id.
+	lockByVolumeID *LockByID
+	server         *controllerServerNoLocked
+}
+
+func (s *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
+	s.lockByName.LockByID(req.GetName())
+	defer s.lockByName.UnlockByID(req.GetName())
+
+	return s.server.CreateVolume(ctx, req)
+}
+
+func (s *controllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVolumeRequest) (*csi.DeleteVolumeResponse, error) {
+	s.lockByVolumeID.LockByID(req.GetVolumeId())
+	defer s.lockByVolumeID.UnlockByID(req.GetVolumeId())
+
+	return s.server.DeleteVolume(ctx, req)
+}
+
+func (s *controllerServer) ValidateVolumeCapabilities(ctx context.Context, req *csi.ValidateVolumeCapabilitiesRequest) (*csi.ValidateVolumeCapabilitiesResponse, error) {
+	s.lockByVolumeID.LockByID(req.GetVolumeId())
+	defer s.lockByVolumeID.UnlockByID(req.GetVolumeId())
+
+	return s.server.ValidateVolumeCapabilities(ctx, req)
+}
+
+func (s *controllerServer) GetCapacity(ctx context.Context, req *csi.GetCapacityRequest) (*csi.GetCapacityResponse, error) {
+	// This reads kube-apiserver only and even if reads dirty state, it is not harmless.
+	// Therefore, it is unnecessary to take lock.
+	return s.server.GetCapacity(ctx, req)
+}
+
+func (s *controllerServer) ControllerGetCapabilities(ctx context.Context, req *csi.ControllerGetCapabilitiesRequest) (*csi.ControllerGetCapabilitiesResponse, error) {
+	// This returns constants only, it is unnecessary to take lock.
+	return s.server.ControllerGetCapabilities(ctx, req)
+}
+
+func (s *controllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateSnapshotRequest) (*csi.CreateSnapshotResponse, error) {
+	s.lockByName.LockByID(req.GetName())
+	defer s.lockByName.UnlockByID(req.GetName())
+
+	return s.server.CreateSnapshot(ctx, req)
+}
+
+func (s *controllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteSnapshotRequest) (*csi.DeleteSnapshotResponse, error) {
+	s.lockByVolumeID.LockByID(req.GetSnapshotId())
+	defer s.lockByVolumeID.UnlockByID(req.GetSnapshotId())
+
+	return s.server.DeleteSnapshot(ctx, req)
+}
+
+func (s *controllerServer) ControllerExpandVolume(ctx context.Context, req *csi.ControllerExpandVolumeRequest) (*csi.ControllerExpandVolumeResponse, error) {
+	s.lockByVolumeID.LockByID(req.GetVolumeId())
+	defer s.lockByVolumeID.UnlockByID(req.GetVolumeId())
+
+	return s.server.ControllerExpandVolume(ctx, req)
 }
 
-type controllerService struct {
+// controllerServerNoLocked implements csi.ControllerServer.
+// It does not take any lock, gRPC calls may be interleaved.
+// Therefore, must not use it directly.
+type controllerServerNoLocked struct {
 	csi.UnimplementedControllerServer
 
 	lvService   *k8s.LogicalVolumeService
 	nodeService *k8s.NodeService
 }
 
-func (s controllerService) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
+func (s controllerServerNoLocked) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
 	capabilities := req.GetVolumeCapabilities()
 	source := req.GetVolumeContentSource()
 	deviceClass := req.GetParameters()[topolvm.GetDeviceClassKey()]
@@ -219,7 +299,7 @@ func (s controllerService) CreateVolume(ctx context.Context, req *csi.CreateVolu
 }
 
 // validateContentSource checks if the request has a data source and returns source volume information.
-func (s controllerService) validateContentSource(ctx context.Context, req *csi.CreateVolumeRequest) (*v1.LogicalVolume, string, error) {
+func (s controllerServerNoLocked) validateContentSource(ctx context.Context, req *csi.CreateVolumeRequest) (*v1.LogicalVolume, string, error) {
 	volumeSource := req.VolumeContentSource
 
 	switch volumeSource.Type.(type) {
@@ -257,7 +337,7 @@ func (s controllerService) validateContentSource(ctx context.Context, req *csi.C
 }
 
 // CreateSnapshot creates a logical volume snapshot.
-func (s controllerService) CreateSnapshot(ctx context.Context, req *csi.CreateSnapshotRequest) (*csi.CreateSnapshotResponse, error) {
+func (s controllerServerNoLocked) CreateSnapshot(ctx context.Context, req *csi.CreateSnapshotRequest) (*csi.CreateSnapshotResponse, error) {
 	// Since the kubernetes snapshots are Read-Only, we set accessType as 'ro' to activate thin-snapshots as read-only volumes
 	accessType := "ro"
 
@@ -314,7 +394,7 @@ func (s controllerService) CreateSnapshot(ctx context.Context, req *csi.CreateSn
 }
 
 // DeleteSnapshot deletes an existing logical volume snapshot.
-func (s controllerService) DeleteSnapshot(ctx context.Context, req *csi.DeleteSnapshotRequest) (*csi.DeleteSnapshotResponse, error) {
+func (s controllerServerNoLocked) DeleteSnapshot(ctx context.Context, req *csi.DeleteSnapshotRequest) (*csi.DeleteSnapshotResponse, error) {
 	ctrlLogger.Info("DeleteSnapshot called",
 		"snapshot_id", req.GetSnapshotId(),
 		"num_secrets", len(req.GetSecrets()))
@@ -355,7 +435,7 @@ func convertRequestCapacity(requestBytes, limitBytes int64) (int64, error) {
 	return (requestBytes-1)>>30 + 1, nil
 }
 
-func (s controllerService) DeleteVolume(ctx context.Context, req *csi.DeleteVolumeRequest) (*csi.DeleteVolumeResponse, error) {
+func (s controllerServerNoLocked) DeleteVolume(ctx context.Context, req *csi.DeleteVolumeRequest) (*csi.DeleteVolumeResponse, error) {
 	ctrlLogger.Info("DeleteVolume called",
 		"volume_id", req.GetVolumeId(),
 		"num_secrets", len(req.GetSecrets()))
@@ -376,7 +456,7 @@ func (s controllerService) DeleteVolume(ctx context.Context, req *csi.DeleteVolu
 	return &csi.DeleteVolumeResponse{}, nil
 }
 
-func (s controllerService) ValidateVolumeCapabilities(ctx context.Context, req *csi.ValidateVolumeCapabilitiesRequest) (*csi.ValidateVolumeCapabilitiesResponse, error) {
+func (s controllerServerNoLocked) ValidateVolumeCapabilities(ctx context.Context, req *csi.ValidateVolumeCapabilitiesRequest) (*csi.ValidateVolumeCapabilitiesResponse, error) {
 	ctrlLogger.Info("ValidateVolumeCapabilities called",
 		"volume_id", req.GetVolumeId(),
 		"volume_context", req.GetVolumeContext(),
@@ -410,7 +490,7 @@ func (s controllerService) ValidateVolumeCapabilities(ctx context.Context, req *
 	}, nil
 }
 
-func (s controllerService) GetCapacity(ctx context.Context, req *csi.GetCapacityRequest) (*csi.GetCapacityResponse, error) {
+func (s controllerServerNoLocked) GetCapacity(ctx context.Context, req *csi.GetCapacityRequest) (*csi.GetCapacityResponse, error) {
 	topology := req.GetAccessibleTopology()
 	capabilities := req.GetVolumeCapabilities()
 	ctrlLogger.Info("GetCapacity called",
@@ -458,7 +538,7 @@ func (s controllerService) GetCapacity(ctx context.Context, req *csi.GetCapacity
 	}, nil
 }
 
-func (s controllerService) ControllerGetCapabilities(context.Context, *csi.ControllerGetCapabilitiesRequest) (*csi.ControllerGetCapabilitiesResponse, error) {
+func (s controllerServerNoLocked) ControllerGetCapabilities(context.Context, *csi.ControllerGetCapabilitiesRequest) (*csi.ControllerGetCapabilitiesResponse, error) {
 	capabilities := []csi.ControllerServiceCapability_RPC_Type{
 		csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
 		csi.ControllerServiceCapability_RPC_CLONE_VOLUME,
@@ -483,7 +563,7 @@ func (s controllerService) ControllerGetCapabilities(context.Context, *csi.Contr
 	}, nil
 }
 
-func (s controllerService) ControllerExpandVolume(ctx context.Context, req *csi.ControllerExpandVolumeRequest) (*csi.ControllerExpandVolumeResponse, error) {
+func (s controllerServerNoLocked) ControllerExpandVolume(ctx context.Context, req *csi.ControllerExpandVolumeRequest) (*csi.ControllerExpandVolumeResponse, error) {
 	volumeID := req.GetVolumeId()
 	ctrlLogger.Info("ControllerExpandVolume called",
 		"volumeID", volumeID,
diff --git a/driver/identity.go b/driver/identity.go
index 28060a6d..410b5f57 100644
--- a/driver/identity.go
+++ b/driver/identity.go
@@ -13,23 +13,23 @@ import (
 
 var idLogger = ctrl.Log.WithName("driver").WithName("identity")
 
-// NewIdentityService returns a new IdentityServer.
+// NewIdentityServer returns a new IdentityServer.
 //
 // ready is a function to check the plugin status.
 // It should return non-nil error if the plugin is not healthy.
 // If the plugin is not yet ready, it should return (false, nil).
 // Otherwise, return (true, nil).
-func NewIdentityService(ready func() (bool, error)) csi.IdentityServer {
-	return &identityService{ready: ready}
+func NewIdentityServer(ready func() (bool, error)) csi.IdentityServer {
+	return &identityServer{ready: ready}
 }
 
-type identityService struct {
+type identityServer struct {
 	csi.UnimplementedIdentityServer
 
 	ready func() (bool, error)
 }
 
-func (s identityService) GetPluginInfo(ctx context.Context, req *csi.GetPluginInfoRequest) (*csi.GetPluginInfoResponse, error) {
+func (s identityServer) GetPluginInfo(ctx context.Context, req *csi.GetPluginInfoRequest) (*csi.GetPluginInfoResponse, error) {
 	idLogger.Info("GetPluginInfo", "req", req.String())
 	return &csi.GetPluginInfoResponse{
 		Name:          topolvm.GetPluginName(),
@@ -37,7 +37,7 @@ func (s identityService) GetPluginInfo(ctx context.Context, req *csi.GetPluginIn
 	}, nil
 }
 
-func (s identityService) GetPluginCapabilities(ctx context.Context, req *csi.GetPluginCapabilitiesRequest) (*csi.GetPluginCapabilitiesResponse, error) {
+func (s identityServer) GetPluginCapabilities(ctx context.Context, req *csi.GetPluginCapabilitiesRequest) (*csi.GetPluginCapabilitiesResponse, error) {
 	idLogger.Info("GetPluginCapabilities", "req", req.String())
 	return &csi.GetPluginCapabilitiesResponse{
 		Capabilities: []*csi.PluginCapability{
@@ -73,7 +73,7 @@ func (s identityService) GetPluginCapabilities(ctx context.Context, req *csi.Get
 	}, nil
 }
 
-func (s identityService) Probe(ctx context.Context, req *csi.ProbeRequest) (*csi.ProbeResponse, error) {
+func (s identityServer) Probe(ctx context.Context, req *csi.ProbeRequest) (*csi.ProbeResponse, error) {
 	idLogger.Info("Probe", "req", req.String())
 	ok, err := s.ready()
 	if err != nil {
diff --git a/driver/k8s/logicalvolume_service.go b/driver/internal/k8s/logicalvolume_service.go
similarity index 99%
rename from driver/k8s/logicalvolume_service.go
rename to driver/internal/k8s/logicalvolume_service.go
index 05d71291..e3cd161f 100644
--- a/driver/k8s/logicalvolume_service.go
+++ b/driver/internal/k8s/logicalvolume_service.go
@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"sync"
 	"time"
 
 	"github.com/topolvm/topolvm"
@@ -26,6 +25,7 @@ import (
 var ErrVolumeNotFound = errors.New("VolumeID is not found")
 
 // LogicalVolumeService represents service for LogicalVolume.
+// This is not concurrent safe, must take lock on caller.
 type LogicalVolumeService struct {
 	writer interface {
 		client.Writer
@@ -33,7 +33,6 @@ type LogicalVolumeService struct {
 	}
 	getter       getter.Interface
 	volumeGetter *volumeGetter
-	mu           sync.Mutex
 }
 
 const (
@@ -153,8 +152,6 @@ func NewLogicalVolumeService(mgr manager.Manager) (*LogicalVolumeService, error)
 // CreateVolume creates volume
 func (s *LogicalVolumeService) CreateVolume(ctx context.Context, node, dc, oc, name, sourceName string, requestGb int64) (string, error) {
 	logger.Info("k8s.CreateVolume called", "name", name, "node", node, "size_gb", requestGb, "sourceName", sourceName)
-	s.mu.Lock()
-	defer s.mu.Unlock()
 	var lv *topolvmv1.LogicalVolume
 	// if the create volume request has no source, proceed with regular lv creation.
 	if sourceName == "" {
@@ -304,8 +301,6 @@ func (s *LogicalVolumeService) CreateSnapshot(ctx context.Context, node, dc, sou
 // ExpandVolume expands volume
 func (s *LogicalVolumeService) ExpandVolume(ctx context.Context, volumeID string, requestGb int64) error {
 	logger.Info("k8s.ExpandVolume called", "volumeID", volumeID, "requestGb", requestGb)
-	s.mu.Lock()
-	defer s.mu.Unlock()
 
 	lv, err := s.GetVolume(ctx, volumeID)
 	if err != nil {
diff --git a/driver/k8s/node_service.go b/driver/internal/k8s/node_service.go
similarity index 100%
rename from driver/k8s/node_service.go
rename to driver/internal/k8s/node_service.go
diff --git a/driver/lock.go b/driver/lock.go
new file mode 100644
index 00000000..786e5999
--- /dev/null
+++ b/driver/lock.go
@@ -0,0 +1,49 @@
+package driver
+
+import (
+	"fmt"
+	"sync"
+)
+
+// LockByID is a mutex which takes lock with a given ID.
+// If same id is given, only one process can take lock.
+// If different id is given, each process can take lock separately.
+type LockByID struct {
+	cond   sync.Cond
+	locked map[string]struct{}
+}
+
+func NewLockWithID() *LockByID {
+	return &LockByID{
+		cond:   *sync.NewCond(&sync.Mutex{}),
+		locked: map[string]struct{}{},
+	}
+}
+
+func (l *LockByID) LockByID(id string) {
+	l.cond.L.Lock()
+	defer l.cond.L.Unlock()
+
+	for {
+		if _, locked := l.locked[id]; !locked {
+			// we can take lock
+			l.locked[id] = struct{}{}
+			return
+		}
+
+		l.cond.Wait()
+	}
+}
+
+func (l *LockByID) UnlockByID(id string) {
+	l.cond.L.Lock()
+	defer l.cond.L.Unlock()
+
+	if _, locked := l.locked[id]; !locked {
+		panic(fmt.Sprintf("BUG: try to unlock not taken lock, id=%s", id))
+	}
+
+	delete(l.locked, id)
+
+	l.cond.Broadcast()
+}
diff --git a/driver/node.go b/driver/node.go
index 78bf3d7d..ca29e79e 100644
--- a/driver/node.go
+++ b/driver/node.go
@@ -11,7 +11,7 @@ import (
 
 	"github.com/topolvm/topolvm"
 	"github.com/topolvm/topolvm/csi"
-	"github.com/topolvm/topolvm/driver/k8s"
+	"github.com/topolvm/topolvm/driver/internal/k8s"
 	"github.com/topolvm/topolvm/filesystem"
 	"github.com/topolvm/topolvm/lvmd/proto"
 	"golang.org/x/sys/unix"
@@ -21,48 +21,103 @@ import (
 	mountutil "k8s.io/mount-utils"
 	utilexec "k8s.io/utils/exec"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 )
 
 const (
 	// DeviceDirectory is a directory where TopoLVM Node service creates device files.
 	DeviceDirectory = "/dev/topolvm"
 
-	mkfsCmd          = "/sbin/mkfs"
-	mountCmd         = "/bin/mount"
-	mountpointCmd    = "/bin/mountpoint"
-	umountCmd        = "/bin/umount"
 	findmntCmd       = "/bin/findmnt"
 	devicePermission = 0600 | unix.S_IFBLK
 )
 
 var nodeLogger = ctrl.Log.WithName("driver").WithName("node")
 
-// NewNodeService returns a new NodeServer.
-func NewNodeService(nodeName string, conn *grpc.ClientConn, service *k8s.LogicalVolumeService) csi.NodeServer {
-	return &nodeService{
-		nodeName:     nodeName,
-		client:       proto.NewVGServiceClient(conn),
-		lvService:    proto.NewLVServiceClient(conn),
-		k8sLVService: service,
-		mounter: mountutil.SafeFormatAndMount{
-			Interface: mountutil.New(""),
-			Exec:      utilexec.New(),
-		},
+// NewNodeServer returns a new NodeServer.
+func NewNodeServer(nodeName string, conn *grpc.ClientConn, mgr manager.Manager) (csi.NodeServer, error) {
+	lvService, err := k8s.NewLogicalVolumeService(mgr)
+	if err != nil {
+		return nil, err
 	}
+
+	return &nodeServer{
+		server: &nodeServerNoLocked{
+			nodeName:     nodeName,
+			client:       proto.NewVGServiceClient(conn),
+			lvService:    proto.NewLVServiceClient(conn),
+			k8sLVService: lvService,
+			mounter: mountutil.SafeFormatAndMount{
+				Interface: mountutil.New(""),
+				Exec:      utilexec.New(),
+			},
+		},
+	}, nil
+}
+
+// This is a wrapper for nodeServerNoLocked to protect concurrent method calls.
+type nodeServer struct {
+	csi.UnimplementedNodeServer
+
+	// This protects concurrent nodeServerNoLocked method calls.
+	// We use a global lock because it assumes that each method does not take a long time,
+	// and we scare about wired behaviors from concurrent device or filesystem operations.
+	mu     sync.Mutex
+	server *nodeServerNoLocked
+}
+
+func (s *nodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	return s.server.NodePublishVolume(ctx, req)
+}
+
+func (s *nodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpublishVolumeRequest) (*csi.NodeUnpublishVolumeResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	return s.server.NodeUnpublishVolume(ctx, req)
+}
+
+func (s *nodeServer) NodeGetVolumeStats(ctx context.Context, req *csi.NodeGetVolumeStatsRequest) (*csi.NodeGetVolumeStatsResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	return s.server.NodeGetVolumeStats(ctx, req)
+}
+
+func (s *nodeServer) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandVolumeRequest) (*csi.NodeExpandVolumeResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	return s.server.NodeExpandVolume(ctx, req)
+}
+
+func (s *nodeServer) NodeGetCapabilities(ctx context.Context, req *csi.NodeGetCapabilitiesRequest) (*csi.NodeGetCapabilitiesResponse, error) {
+	// This returns constant value only, it is unnecessary to take lock.
+	return s.server.NodeGetCapabilities(ctx, req)
 }
 
-type nodeService struct {
+func (s *nodeServer) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoRequest) (*csi.NodeGetInfoResponse, error) {
+	// This returns unmodified value only, it is unnecessary to take lock.
+	return s.server.NodeGetInfo(ctx, req)
+}
+
+// nodeServerNoLocked implements csi.NodeServer.
+// It does not take any lock, gRPC calls may be interleaved.
+// Therefore, must not use it directly.
+type nodeServerNoLocked struct {
 	csi.UnimplementedNodeServer
 
 	nodeName     string
 	client       proto.VGServiceClient
 	lvService    proto.LVServiceClient
 	k8sLVService *k8s.LogicalVolumeService
-	mu           sync.Mutex
 	mounter      mountutil.SafeFormatAndMount
 }
 
-func (s *nodeService) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) {
+func (s *nodeServerNoLocked) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) {
 	volumeContext := req.GetVolumeContext()
 	volumeID := req.GetVolumeId()
 
@@ -89,9 +144,14 @@ func (s *nodeService) NodePublishVolume(ctx context.Context, req *csi.NodePublis
 	if !(isBlockVol || isFsVol) {
 		return nil, status.Errorf(codes.InvalidArgument, "no supported volume capability: %v", req.GetVolumeCapability())
 	}
-
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	// we only support SINGLE_NODE_WRITER
+	accessMode := req.GetVolumeCapability().GetAccessMode().GetMode()
+	switch accessMode {
+	case csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER:
+	default:
+		modeName := csi.VolumeCapability_AccessMode_Mode_name[int32(accessMode)]
+		return nil, status.Errorf(codes.FailedPrecondition, "unsupported access mode: %s (%d)", modeName, accessMode)
+	}
 
 	var lv *proto.LogicalVolume
 	var err error
@@ -119,20 +179,12 @@ func (s *nodeService) NodePublishVolume(ctx context.Context, req *csi.NodePublis
 	return &csi.NodePublishVolumeResponse{}, nil
 }
 
-func (s *nodeService) nodePublishFilesystemVolume(req *csi.NodePublishVolumeRequest, lv *proto.LogicalVolume) error {
+func (s *nodeServerNoLocked) nodePublishFilesystemVolume(req *csi.NodePublishVolumeRequest, lv *proto.LogicalVolume) error {
 	// Check request
 	mountOption := req.GetVolumeCapability().GetMount()
 	if mountOption.FsType == "" {
 		mountOption.FsType = "ext4"
 	}
-	// we only support SINGLE_NODE_WRITER
-	accessMode := req.GetVolumeCapability().GetAccessMode().GetMode()
-	switch accessMode {
-	case csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER:
-	default:
-		modeName := csi.VolumeCapability_AccessMode_Mode_name[int32(accessMode)]
-		return status.Errorf(codes.FailedPrecondition, "unsupported access mode: %s (%d)", modeName, accessMode)
-	}
 
 	// Find lv and create a block device with it
 	device := filepath.Join(DeviceDirectory, req.GetVolumeId())
@@ -194,7 +246,7 @@ func (s *nodeService) nodePublishFilesystemVolume(req *csi.NodePublishVolumeRequ
 	return nil
 }
 
-func (s *nodeService) createDeviceIfNeeded(device string, lv *proto.LogicalVolume) error {
+func (s *nodeServerNoLocked) createDeviceIfNeeded(device string, lv *proto.LogicalVolume) error {
 	var stat unix.Stat_t
 	err := filesystem.Stat(device, &stat)
 	switch err {
@@ -209,6 +261,11 @@ func (s *nodeService) createDeviceIfNeeded(device string, lv *proto.LogicalVolum
 		}
 		fallthrough
 	case unix.ENOENT:
+		err = os.MkdirAll(path.Dir(device), 0755)
+		if err != nil {
+			return status.Errorf(codes.Internal, "mkdir failed: target=%s, error=%v", path.Dir(device), err)
+		}
+
 		devno := unix.Mkdev(lv.DevMajor, lv.DevMinor)
 		if err := filesystem.Mknod(device, devicePermission, int(devno)); err != nil {
 			return status.Errorf(codes.Internal, "mknod failed for %s. major=%d, minor=%d, error=%v",
@@ -220,41 +277,21 @@ func (s *nodeService) createDeviceIfNeeded(device string, lv *proto.LogicalVolum
 	return nil
 }
 
-func (s *nodeService) nodePublishBlockVolume(req *csi.NodePublishVolumeRequest, lv *proto.LogicalVolume) error {
+func (s *nodeServerNoLocked) nodePublishBlockVolume(req *csi.NodePublishVolumeRequest, lv *proto.LogicalVolume) error {
 	// Find lv and create a block device with it
-	var stat unix.Stat_t
-	target := req.GetTargetPath()
-	err := filesystem.Stat(target, &stat)
-	switch err {
-	case nil:
-		if stat.Rdev == unix.Mkdev(lv.DevMajor, lv.DevMinor) && stat.Mode&devicePermission == devicePermission {
-			return nil
-		}
-		if err := os.Remove(target); err != nil {
-			return status.Errorf(codes.Internal, "failed to remove %s", target)
-		}
-	case unix.ENOENT:
-	default:
-		return status.Errorf(codes.Internal, "failed to stat: %v", err)
-	}
-
-	err = os.MkdirAll(path.Dir(target), 0755)
+	targetPath := req.GetTargetPath()
+	err := s.createDeviceIfNeeded(targetPath, lv)
 	if err != nil {
-		return status.Errorf(codes.Internal, "mkdir failed: target=%s, error=%v", path.Dir(target), err)
-	}
-
-	devno := unix.Mkdev(lv.DevMajor, lv.DevMinor)
-	if err := filesystem.Mknod(target, devicePermission, int(devno)); err != nil {
-		return status.Errorf(codes.Internal, "mknod failed for %s: error=%v", target, err)
+		return err
 	}
 
 	nodeLogger.Info("NodePublishVolume(block) succeeded",
 		"volume_id", req.GetVolumeId(),
-		"target_path", target)
+		"target_path", targetPath)
 	return nil
 }
 
-func (s *nodeService) findVolumeByID(listResp *proto.GetLVListResponse, name string) *proto.LogicalVolume {
+func (s *nodeServerNoLocked) findVolumeByID(listResp *proto.GetLVListResponse, name string) *proto.LogicalVolume {
 	for _, v := range listResp.Volumes {
 		if v.Name == name {
 			return v
@@ -263,7 +300,7 @@ func (s *nodeService) findVolumeByID(listResp *proto.GetLVListResponse, name str
 	return nil
 }
 
-func (s *nodeService) getLvFromContext(ctx context.Context, deviceClass, volumeID string) (*proto.LogicalVolume, error) {
+func (s *nodeServerNoLocked) getLvFromContext(ctx context.Context, deviceClass, volumeID string) (*proto.LogicalVolume, error) {
 	listResp, err := s.client.GetLVList(ctx, &proto.GetLVListRequest{DeviceClass: deviceClass})
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, "failed to list LV: %v", err)
@@ -271,112 +308,110 @@ func (s *nodeService) getLvFromContext(ctx context.Context, deviceClass, volumeI
 	return s.findVolumeByID(listResp, volumeID), nil
 }
 
-func (s *nodeService) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpublishVolumeRequest) (*csi.NodeUnpublishVolumeResponse, error) {
-	volID := req.GetVolumeId()
-	target := req.GetTargetPath()
+func (s *nodeServerNoLocked) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpublishVolumeRequest) (*csi.NodeUnpublishVolumeResponse, error) {
+	volumeID := req.GetVolumeId()
+	targetPath := req.GetTargetPath()
 	nodeLogger.Info("NodeUnpublishVolume called",
-		"volume_id", volID,
-		"target_path", target)
+		"volume_id", volumeID,
+		"target_path", targetPath)
 
-	if len(volID) == 0 {
+	if len(volumeID) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "no volume_id is provided")
 	}
-	if len(target) == 0 {
+	if len(targetPath) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "no target_path is provided")
 	}
 
-	s.mu.Lock()
-	defer s.mu.Unlock()
-
-	device := filepath.Join(DeviceDirectory, volID)
+	device := filepath.Join(DeviceDirectory, volumeID)
 
-	info, err := os.Stat(target)
+	info, err := os.Stat(targetPath)
 	if os.IsNotExist(err) {
 		// target_path does not exist, but device for mount-type PV may still exist.
 		_ = os.Remove(device)
 		return &csi.NodeUnpublishVolumeResponse{}, nil
 	} else if err != nil {
-		return nil, status.Errorf(codes.Internal, "stat failed for %s: %v", target, err)
+		return nil, status.Errorf(codes.Internal, "stat failed for %s: %v", targetPath, err)
 	}
 
 	// remove device file if target_path is device, unmount target_path otherwise
 	if info.IsDir() {
-		unpublishResp, err := s.nodeUnpublishFilesystemVolume(req, device)
-		if err != nil {
-			return unpublishResp, err
-		}
-		return &csi.NodeUnpublishVolumeResponse{}, nil
+		err = s.nodeUnpublishFilesystemVolume(req, device)
+	} else {
+		err = s.nodeUnpublishBlockVolume(req)
+	}
+	if err != nil {
+		return nil, err
 	}
-	return s.nodeUnpublishBlockVolume(req)
+	return &csi.NodeUnpublishVolumeResponse{}, nil
 }
 
-func (s *nodeService) nodeUnpublishFilesystemVolume(req *csi.NodeUnpublishVolumeRequest, device string) (*csi.NodeUnpublishVolumeResponse, error) {
-	target := req.GetTargetPath()
+func (s *nodeServerNoLocked) nodeUnpublishFilesystemVolume(req *csi.NodeUnpublishVolumeRequest, device string) error {
+	targetPath := req.GetTargetPath()
 
-	mounted, err := filesystem.IsMounted(device, target)
+	mounted, err := filesystem.IsMounted(device, targetPath)
 	if err != nil {
-		return nil, status.Errorf(codes.Internal, "mount check failed: target=%s, error=%v", target, err)
+		return status.Errorf(codes.Internal, "mount check failed: target=%s, error=%v", targetPath, err)
 	}
 	if mounted {
-		if err := s.mounter.Unmount(target); err != nil {
-			return nil, status.Errorf(codes.Internal, "unmount failed for %s: error=%v", target, err)
+		if err := s.mounter.Unmount(targetPath); err != nil {
+			return status.Errorf(codes.Internal, "unmount failed for %s: error=%v", targetPath, err)
 		}
 	}
 
-	if err := os.RemoveAll(target); err != nil {
-		return nil, status.Errorf(codes.Internal, "remove dir failed for %s: error=%v", target, err)
+	if err := os.RemoveAll(targetPath); err != nil {
+		return status.Errorf(codes.Internal, "remove dir failed for %s: error=%v", targetPath, err)
 	}
 
 	err = os.Remove(device)
 	if err != nil && !os.IsNotExist(err) {
-		return nil, status.Errorf(codes.Internal, "remove device failed for %s: error=%v", device, err)
+		return status.Errorf(codes.Internal, "remove device failed for %s: error=%v", device, err)
 	}
 
 	nodeLogger.Info("NodeUnpublishVolume(fs) is succeeded",
 		"volume_id", req.GetVolumeId(),
-		"target_path", target)
-	return &csi.NodeUnpublishVolumeResponse{}, nil
+		"target_path", targetPath)
+	return nil
 }
 
-func (s *nodeService) nodeUnpublishBlockVolume(req *csi.NodeUnpublishVolumeRequest) (*csi.NodeUnpublishVolumeResponse, error) {
+func (s *nodeServerNoLocked) nodeUnpublishBlockVolume(req *csi.NodeUnpublishVolumeRequest) error {
 	if err := os.Remove(req.GetTargetPath()); err != nil {
-		return nil, status.Errorf(codes.Internal, "remove failed for %s: error=%v", req.GetTargetPath(), err)
+		return status.Errorf(codes.Internal, "remove failed for %s: error=%v", req.GetTargetPath(), err)
 	}
 	nodeLogger.Info("NodeUnpublishVolume(block) is succeeded",
 		"volume_id", req.GetVolumeId(),
 		"target_path", req.GetTargetPath())
-	return &csi.NodeUnpublishVolumeResponse{}, nil
+	return nil
 }
 
-func (s *nodeService) NodeGetVolumeStats(ctx context.Context, req *csi.NodeGetVolumeStatsRequest) (*csi.NodeGetVolumeStatsResponse, error) {
-	volID := req.GetVolumeId()
-	p := req.GetVolumePath()
-	nodeLogger.Info("NodeGetVolumeStats is called", "volume_id", volID, "volume_path", p)
-	if len(volID) == 0 {
+func (s *nodeServerNoLocked) NodeGetVolumeStats(ctx context.Context, req *csi.NodeGetVolumeStatsRequest) (*csi.NodeGetVolumeStatsResponse, error) {
+	volumeID := req.GetVolumeId()
+	volumePath := req.GetVolumePath()
+	nodeLogger.Info("NodeGetVolumeStats is called", "volume_id", volumeID, "volume_path", volumePath)
+	if len(volumeID) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "no volume_id is provided")
 	}
-	if len(p) == 0 {
+	if len(volumePath) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "no volume_path is provided")
 	}
 
 	var st unix.Stat_t
-	switch err := filesystem.Stat(p, &st); err {
+	switch err := filesystem.Stat(volumePath, &st); err {
 	case unix.ENOENT:
-		return nil, status.Error(codes.NotFound, "Volume is not found at "+p)
+		return nil, status.Error(codes.NotFound, "Volume is not found at "+volumePath)
 	case nil:
 	default:
-		return nil, status.Errorf(codes.Internal, "stat on %s was failed: %v", p, err)
+		return nil, status.Errorf(codes.Internal, "stat on %s was failed: %v", volumePath, err)
 	}
 
 	if (st.Mode & unix.S_IFMT) == unix.S_IFBLK {
-		f, err := os.Open(p)
+		f, err := os.Open(volumePath)
 		if err != nil {
-			return nil, status.Errorf(codes.Internal, "open on %s was failed: %v", p, err)
+			return nil, status.Errorf(codes.Internal, "open on %s was failed: %v", volumePath, err)
 		}
 		defer f.Close()
 		pos, err := f.Seek(0, io.SeekEnd)
 		if err != nil {
-			return nil, status.Errorf(codes.Internal, "seek on %s was failed: %v", p, err)
+			return nil, status.Errorf(codes.Internal, "seek on %s was failed: %v", volumePath, err)
 		}
 		return &csi.NodeGetVolumeStatsResponse{
 			Usage: []*csi.VolumeUsage{{Total: pos, Unit: csi.VolumeUsage_BYTES}},
@@ -384,12 +419,12 @@ func (s *nodeService) NodeGetVolumeStats(ctx context.Context, req *csi.NodeGetVo
 	}
 
 	if st.Mode&unix.S_IFDIR == 0 {
-		return nil, status.Errorf(codes.Internal, "invalid mode bits for %s: %d", p, st.Mode)
+		return nil, status.Errorf(codes.Internal, "invalid mode bits for %s: %d", volumePath, st.Mode)
 	}
 
 	var sfs unix.Statfs_t
-	if err := filesystem.Statfs(p, &sfs); err != nil {
-		return nil, status.Errorf(codes.Internal, "statvfs on %s was failed: %v", p, err)
+	if err := filesystem.Statfs(volumePath, &sfs); err != nil {
+		return nil, status.Errorf(codes.Internal, "statfs on %s was failed: %v", volumePath, err)
 	}
 
 	var usage []*csi.VolumeUsage
@@ -412,21 +447,21 @@ func (s *nodeService) NodeGetVolumeStats(ctx context.Context, req *csi.NodeGetVo
 	return &csi.NodeGetVolumeStatsResponse{Usage: usage}, nil
 }
 
-func (s *nodeService) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandVolumeRequest) (*csi.NodeExpandVolumeResponse, error) {
-	vid := req.GetVolumeId()
-	vpath := req.GetVolumePath()
+func (s *nodeServerNoLocked) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandVolumeRequest) (*csi.NodeExpandVolumeResponse, error) {
+	volumeID := req.GetVolumeId()
+	volumePath := req.GetVolumePath()
 
 	nodeLogger.Info("NodeExpandVolume is called",
-		"volume_id", vid,
-		"volume_path", vpath,
+		"volume_id", volumeID,
+		"volume_path", volumePath,
 		"required", req.GetCapacityRange().GetRequiredBytes(),
 		"limit", req.GetCapacityRange().GetLimitBytes(),
 	)
 
-	if len(vid) == 0 {
+	if len(volumeID) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "no volume_id is provided")
 	}
-	if len(vpath) == 0 {
+	if len(volumePath) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "no volume_path is provided")
 	}
 
@@ -439,37 +474,37 @@ func (s *nodeService) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandV
 
 	// Device type (block or fs, fs type detection) checking will be removed after CSI v1.2.0
 	// because `volume_capability` field will be added in csi.NodeExpandVolumeRequest
-	info, err := os.Stat(vpath)
+	info, err := os.Stat(volumePath)
 	if err != nil {
 		if os.IsNotExist(err) {
-			return nil, status.Errorf(codes.NotFound, "volume path is not exist: %s", vpath)
+			return nil, status.Errorf(codes.NotFound, "volume path is not exist: %s", volumePath)
 		}
-		return nil, status.Errorf(codes.Internal, "stat failed for %s: %v", vpath, err)
+		return nil, status.Errorf(codes.Internal, "stat failed for %s: %v", volumePath, err)
 	}
 
 	isBlock := !info.IsDir()
 	if isBlock {
 		nodeLogger.Info("NodeExpandVolume(block) is skipped",
-			"volume_id", vid,
-			"target_path", vpath,
+			"volume_id", volumeID,
+			"target_path", volumePath,
 		)
 		return &csi.NodeExpandVolumeResponse{}, nil
 	}
 
-	device := filepath.Join(DeviceDirectory, vid)
-	lvr, err := s.k8sLVService.GetVolume(ctx, vid)
+	device := filepath.Join(DeviceDirectory, volumeID)
+	lvr, err := s.k8sLVService.GetVolume(ctx, volumeID)
 	deviceClass := topolvm.DefaultDeviceClassName
 	if err == nil {
 		deviceClass = lvr.Spec.DeviceClass
 	} else if err != k8s.ErrVolumeNotFound {
 		return nil, err
 	}
-	lv, err := s.getLvFromContext(ctx, deviceClass, vid)
+	lv, err := s.getLvFromContext(ctx, deviceClass, volumeID)
 	if err != nil {
 		return nil, err
 	}
 	if lv == nil {
-		return nil, status.Errorf(codes.NotFound, "failed to find LV: %s", vid)
+		return nil, status.Errorf(codes.NotFound, "failed to find LV: %s", volumeID)
 	}
 	err = s.createDeviceIfNeeded(device, lv)
 	if err != nil {
@@ -484,20 +519,17 @@ func (s *nodeService) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandV
 
 	devicePath := strings.TrimSpace(string(output))
 	if len(devicePath) == 0 {
-		return nil, status.Errorf(codes.Internal, "filesystem %s is not mounted at %s", vid, vpath)
+		return nil, status.Errorf(codes.Internal, "filesystem %s is not mounted at %s", volumeID, volumePath)
 	}
 
-	s.mu.Lock()
-	defer s.mu.Unlock()
-
 	r := mountutil.NewResizeFs(s.mounter.Exec)
-	if _, err := r.Resize(device, vpath); err != nil {
-		return nil, status.Errorf(codes.Internal, "failed to resize filesystem %s (mounted at: %s): %v", vid, vpath, err)
+	if _, err := r.Resize(device, volumePath); err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to resize filesystem %s (mounted at: %s): %v", volumeID, volumePath, err)
 	}
 
 	nodeLogger.Info("NodeExpandVolume(fs) is succeeded",
-		"volume_id", vid,
-		"target_path", vpath,
+		"volume_id", volumeID,
+		"target_path", volumePath,
 	)
 
 	// `capacity_bytes` in NodeExpandVolumeResponse is defined as OPTIONAL.
@@ -507,7 +539,7 @@ func (s *nodeService) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandV
 	return &csi.NodeExpandVolumeResponse{}, nil
 }
 
-func (s *nodeService) NodeGetCapabilities(context.Context, *csi.NodeGetCapabilitiesRequest) (*csi.NodeGetCapabilitiesResponse, error) {
+func (s *nodeServerNoLocked) NodeGetCapabilities(context.Context, *csi.NodeGetCapabilitiesRequest) (*csi.NodeGetCapabilitiesResponse, error) {
 	capabilities := []csi.NodeServiceCapability_RPC_Type{
 		csi.NodeServiceCapability_RPC_GET_VOLUME_STATS,
 		csi.NodeServiceCapability_RPC_EXPAND_VOLUME,
@@ -529,7 +561,7 @@ func (s *nodeService) NodeGetCapabilities(context.Context, *csi.NodeGetCapabilit
 	}, nil
 }
 
-func (s *nodeService) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoRequest) (*csi.NodeGetInfoResponse, error) {
+func (s *nodeServerNoLocked) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoRequest) (*csi.NodeGetInfoResponse, error) {
 	return &csi.NodeGetInfoResponse{
 		NodeId: s.nodeName,
 		AccessibleTopology: &csi.Topology{
diff --git a/example/Makefile b/example/Makefile
index 74d34d9f..276c699e 100644
--- a/example/Makefile
+++ b/example/Makefile
@@ -2,7 +2,7 @@ USE_LEGACY ?= ""
 
 KUBERNETES_VERSION := 1.25.3
 HELM_VERSION := 3.10.2
-TOPOLVM_VERSION := 0.18.0
+TOPOLVM_VERSION := 0.17.0
 CERT_MANAGER_VERSION := v1.7.0
 
 KIND_CLUSTER_NAME := topolvm-example
diff --git a/example/README.md b/example/README.md
index efcebb50..d4eab071 100644
--- a/example/README.md
+++ b/example/README.md
@@ -8,7 +8,7 @@ and loopback block devices to run `lvmd`.
 You can try TopoLVM using the example with tag like following command. The main branch is being edited, so it may not work.
 
 ```
-$ git checkout v0.18.0
+$ git checkout v0.17.0
 ```
 
 To start the demonstration environment, run the following commands:
diff --git a/go.mod b/go.mod
index ccf3b182..9466ea45 100644
--- a/go.mod
+++ b/go.mod
@@ -17,7 +17,7 @@ require (
 	github.com/pseudomuto/protoc-gen-doc v1.5.0
 	github.com/spf13/cobra v1.4.0
 	github.com/spf13/viper v1.10.1
-	golang.org/x/sys v0.5.0
+	golang.org/x/sys v0.3.0
 	google.golang.org/grpc v1.48.0
 	google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.2.0
 	google.golang.org/protobuf v1.28.0
@@ -98,10 +98,10 @@ require (
 	go.uber.org/zap v1.21.0 // indirect
 	golang.org/x/crypto v0.4.0 // indirect
 	golang.org/x/mod v0.7.0 // indirect
-	golang.org/x/net v0.7.0 // indirect
+	golang.org/x/net v0.3.0 // indirect
 	golang.org/x/oauth2 v0.0.0-20211104180415-d3ed0bb246c8 // indirect
-	golang.org/x/term v0.5.0 // indirect
-	golang.org/x/text v0.7.0 // indirect
+	golang.org/x/term v0.3.0 // indirect
+	golang.org/x/text v0.5.0 // indirect
 	golang.org/x/time v0.0.0-20220609170525-579cf78fd858 // indirect
 	golang.org/x/tools v0.4.0 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.2.0 // indirect
diff --git a/go.sum b/go.sum
index 42822b0c..f9e0459f 100644
--- a/go.sum
+++ b/go.sum
@@ -552,8 +552,8 @@ golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f/go.mod h1:9nx3DQGgdP8bBQD5qx
 golang.org/x/net v0.0.0-20220225172249-27dd8689420f/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
 golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
 golang.org/x/net v0.0.0-20220802222814-0bcc04d9c69b/go.mod h1:YDH+HFinaLZZlnHAfSS6ZXJJ9M9t4Dl22yv3iI2vPwk=
-golang.org/x/net v0.7.0 h1:rJrUqqhjsgNp7KqAIc25s9pZnjU7TUcSY7HcVZjdn1g=
-golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
+golang.org/x/net v0.3.0 h1:VWL6FNY2bEEmsGVKabSlHu5Irp34xmMRoqb/9lF9lxk=
+golang.org/x/net v0.3.0/go.mod h1:MBQ8lrhLObU/6UmLb4fmbmk5OcyYmqtbGd/9yIeKjEE=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -657,12 +657,12 @@ golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220422013727-9388b58f7150/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220731174439-a90be440212d/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.5.0 h1:MUK/U/4lj1t1oPg0HfuXDN/Z1wv31ZJ/YcPiGccS4DU=
-golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.3.0 h1:w8ZOecv6NaNa/zC8944JTU3vz4u6Lagfk4RPQxv92NQ=
+golang.org/x/sys v0.3.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
-golang.org/x/term v0.5.0 h1:n2a8QNdAb0sZNpU9R1ALUXBbY+w51fCQDN+7EdxNBsY=
-golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
+golang.org/x/term v0.3.0 h1:qoo4akIqOcDME5bhc/NgxUdovd6BSS2uMsVjB56q1xI=
+golang.org/x/term v0.3.0/go.mod h1:q750SLmJuPmVoN1blW3UFBPREJfb1KmY3vwxfr+nFDA=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -672,8 +672,8 @@ golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
-golang.org/x/text v0.7.0 h1:4BRB4x83lYWy72KwLD/qYDuTu7q9PjSagHvijDw7cLo=
-golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/text v0.5.0 h1:OLmvp0KP+FVG99Ct/qFiL/Fhk4zp4QQnZ7b2U+5piUM=
+golang.org/x/text v0.5.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
diff --git a/lvmd/vgservice.go b/lvmd/vgservice.go
index a6aa9e8b..0919d150 100644
--- a/lvmd/vgservice.go
+++ b/lvmd/vgservice.go
@@ -27,6 +27,7 @@ type vgService struct {
 	proto.UnimplementedVGServiceServer
 	dcManager *DeviceClassManager
 
+	// mu protects watcherCounter and watchers. must take it when use them.
 	mu             sync.Mutex
 	watcherCounter int
 	watchers       map[int]chan struct{}
diff --git a/pkg/topolvm-controller/cmd/run.go b/pkg/topolvm-controller/cmd/run.go
index 9a73f56f..63b86689 100644
--- a/pkg/topolvm-controller/cmd/run.go
+++ b/pkg/topolvm-controller/cmd/run.go
@@ -13,7 +13,6 @@ import (
 	"github.com/topolvm/topolvm/controllers"
 	"github.com/topolvm/topolvm/csi"
 	"github.com/topolvm/topolvm/driver"
-	"github.com/topolvm/topolvm/driver/k8s"
 	"github.com/topolvm/topolvm/hook"
 	"github.com/topolvm/topolvm/runners"
 	"google.golang.org/grpc"
@@ -114,15 +113,13 @@ func subMain() error {
 	}
 
 	// Add gRPC server to manager.
-	s, err := k8s.NewLogicalVolumeService(mgr)
+	grpcServer := grpc.NewServer()
+	csi.RegisterIdentityServer(grpcServer, driver.NewIdentityServer(checker.Ready))
+	controllerSever, err := driver.NewControllerServer(mgr)
 	if err != nil {
 		return err
 	}
-	n := k8s.NewNodeService(client)
-
-	grpcServer := grpc.NewServer()
-	csi.RegisterIdentityServer(grpcServer, driver.NewIdentityService(checker.Ready))
-	csi.RegisterControllerServer(grpcServer, driver.NewControllerService(s, n))
+	csi.RegisterControllerServer(grpcServer, controllerSever)
 
 	// gRPC service itself should run even when the manager is *not* a leader
 	// because CSI sidecar containers choose a leader.
diff --git a/pkg/topolvm-node/cmd/run.go b/pkg/topolvm-node/cmd/run.go
index fa2e4f77..c2df56e9 100644
--- a/pkg/topolvm-node/cmd/run.go
+++ b/pkg/topolvm-node/cmd/run.go
@@ -15,7 +15,6 @@ import (
 	"github.com/topolvm/topolvm/controllers"
 	"github.com/topolvm/topolvm/csi"
 	"github.com/topolvm/topolvm/driver"
-	"github.com/topolvm/topolvm/driver/k8s"
 	"github.com/topolvm/topolvm/lvmd/proto"
 	"github.com/topolvm/topolvm/runners"
 	"google.golang.org/grpc"
@@ -95,16 +94,16 @@ func subMain() error {
 	}
 
 	// Add gRPC server to manager.
-	s, err := k8s.NewLogicalVolumeService(mgr)
-	if err != nil {
-		return err
-	}
 	if err := os.MkdirAll(driver.DeviceDirectory, 0755); err != nil {
 		return err
 	}
 	grpcServer := grpc.NewServer(grpc.UnaryInterceptor(ErrorLoggingInterceptor))
-	csi.RegisterIdentityServer(grpcServer, driver.NewIdentityService(checker.Ready))
-	csi.RegisterNodeServer(grpcServer, driver.NewNodeService(nodename, conn, s))
+	csi.RegisterIdentityServer(grpcServer, driver.NewIdentityServer(checker.Ready))
+	nodeServer, err := driver.NewNodeServer(nodename, conn, mgr)
+	if err != nil {
+		return err
+	}
+	csi.RegisterNodeServer(grpcServer, nodeServer)
 	err = mgr.Add(runners.NewGRPCRunner(grpcServer, config.csiSocket, false))
 	if err != nil {
 		return err
diff --git a/runners/checker.go b/runners/checker.go
index dfb0501e..9c0c5653 100644
--- a/runners/checker.go
+++ b/runners/checker.go
@@ -64,7 +64,7 @@ func (c *readinessCheck) NeedLeaderElection() bool {
 	return false
 }
 
-// Ready can be passed to driver.NewIdentityService.
+// Ready can be passed to driver.NewIdentityServer.
 func (c *readinessCheck) Ready() (bool, error) {
 	c.mu.RLock()
 	defer c.mu.RUnlock()


Test patch:
diff --git a/driver/lock_test.go b/driver/lock_test.go
new file mode 100644
index 00000000..baface52
--- /dev/null
+++ b/driver/lock_test.go
@@ -0,0 +1,56 @@
+package driver
+
+import (
+	"testing"
+	"time"
+)
+
+func TestPreventConcurrentExecutionBySameID(t *testing.T) {
+	l := NewLockWithID()
+
+	l.LockByID("a")
+
+	c := make(chan struct{})
+
+	go func() {
+		l.LockByID("a")
+		defer l.UnlockByID("a")
+
+		close(c)
+	}()
+
+	select {
+	case <-c:
+		t.Error("Failed to prevent concurrent execution")
+	case <-time.After(time.Second):
+		// success
+	}
+
+	// check goroutine can progress if the lock is released.
+	l.UnlockByID("a")
+
+	<-c
+}
+
+func TestAllowConcurrentExecutionByDifferentID(t *testing.T) {
+	l := NewLockWithID()
+
+	l.LockByID("a")
+	defer l.UnlockByID("a")
+
+	c := make(chan struct{})
+
+	go func() {
+		l.LockByID("b")
+		defer l.UnlockByID("b")
+
+		close(c)
+	}()
+
+	select {
+	case <-c:
+		// success
+	case <-time.After(time.Second):
+		t.Error("Failed to allow concurrent execution")
+	}
+}