Repository: kubernetes-sigs/gcp-filestore-csi-driver
Pull request #125 (2021-06-10 05:20:56)

Serialize volume operations for a given volume and return appropriate error codes
1. Use mutex for volume ops locking
2. Use appropriate error codes for expected long running ops

<!--  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

**What this PR does / why we need it**:
Filestore operations are long running by default. We need to differentiate expected operations from internal errors to avoid false positives in SLO breach. This PR ensures that for expected filestore operations DeadlineExceeded is returned.
Further, volume locking is implemented in order to serialize volume operations in controller and node.
- For a given volume ID, create/delete/createsnapshot is serialized.
- For a given volume ID, node stage/unstage operation is serialized
- Node publish/unpublish operations are serialized at a targetPath key, since we want to node publish operations to work in parallel, for a given staged volume.

**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**:
Test manually by reducing the caller timeout flags of csi-provisioner and csi-snapshotter sidecar [this](https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/blob/master/deploy/kubernetes/base/controller/controller.yaml#L30) and [this](https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/blob/master/deploy/kubernetes/base/controller/controller.yaml#L90) to very small value like 20 seconds, and checked that appropriate deadline exceeded errors are returned during filestore instance provision/delete and create/delete snapshot ops.
Also tested with deployment of ~50 replicas scheduled on a single node, to ensure node publish work in parallel for each pod.

**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
Serialize volume operations for a given volume and return appropriate error codes
```


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

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/pull/125#" title="Author self-approved">saikat-royc</a>*

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

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

- ~~[OWNERS](https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/blob/master/OWNERS)~~ [saikat-royc]

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":[]} -->
k8s-ci-robot: @saikat-royc: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-gcp-filestore-csi-driver-verify | 1e242fd61eb9c74ca0c2e0a7aa9a120a48f91826 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/125/pull-gcp-filestore-csi-driver-verify/1402858299799375872/) | `/test pull-gcp-filestore-csi-driver-verify`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=125). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<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. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @saikat-royc: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-gcp-filestore-csi-driver-verify | a4f2fb6a4fba60f95b38c64e75f401cbdc13d195 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/125/pull-gcp-filestore-csi-driver-verify/1402861528717201408/) | `/test pull-gcp-filestore-csi-driver-verify`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=125). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<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. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @saikat-royc: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-gcp-filestore-csi-driver-e2e | 44e8361ccd58bafecee3e2735c2c2d5538726427 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/125/pull-gcp-filestore-csi-driver-e2e/1402865370192154624/) | `/test pull-gcp-filestore-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=125). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<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. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
saikat-royc: /retest
saikat-royc: /test pull-gcp-filestore-csi-driver-kubernetes-integration
k8s-ci-robot: @saikat-royc: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-gcp-filestore-csi-driver-kubernetes-integration | 44e8361ccd58bafecee3e2735c2c2d5538726427 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/125/pull-gcp-filestore-csi-driver-kubernetes-integration/1402867141354459136/) | `/test pull-gcp-filestore-csi-driver-kubernetes-integration`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=125). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<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. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
saikat-royc: /hold 'pull-gcp-filestore-csi-driver-kubernetes-integration' v1beta1 csidriver not recognized in 1.21
saikat-royc: /test pull-gcp-filestore-csi-driver-kubernetes-integration
saikat-royc: /hold cancel
saikat-royc: /test pull-gcp-filestore-csi-driver-kubernetes-integration
k8s-ci-robot: @saikat-royc: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-gcp-filestore-csi-driver-kubernetes-integration | 8516364ebadda1ea82fdeb170527c23af9fea868 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/125/pull-gcp-filestore-csi-driver-kubernetes-integration/1405343350696972288/) | `/test pull-gcp-filestore-csi-driver-kubernetes-integration`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=125). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<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. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
saikat-royc: /retest
k8s-ci-robot: @saikat-royc: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-gcp-filestore-csi-driver-kubernetes-integration | 8516364ebadda1ea82fdeb170527c23af9fea868 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/125/pull-gcp-filestore-csi-driver-kubernetes-integration/1405381951774265344/) | `/test pull-gcp-filestore-csi-driver-kubernetes-integration`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=125). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<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. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
saikat-royc: /retest
k8s-ci-robot: @saikat-royc: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-gcp-filestore-csi-driver-kubernetes-integration | 8516364ebadda1ea82fdeb170527c23af9fea868 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/125/pull-gcp-filestore-csi-driver-kubernetes-integration/1405556308853133312/) | `/test pull-gcp-filestore-csi-driver-kubernetes-integration`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=125). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<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. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @saikat-royc: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-gcp-filestore-csi-driver-kubernetes-integration | 8516364ebadda1ea82fdeb170527c23af9fea868 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/125/pull-gcp-filestore-csi-driver-kubernetes-integration/1405556308853133312/) | `/test pull-gcp-filestore-csi-driver-kubernetes-integration`
pull-gcp-filestore-csi-driver-e2e | 3c59c5bef332cbab041d52e11d7fc7a961cbee56 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/125/pull-gcp-filestore-csi-driver-e2e/1405594730976251904/) | `/test pull-gcp-filestore-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=125). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<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. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
saikat-royc: /retest
k8s-ci-robot: @saikat-royc: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-gcp-filestore-csi-driver-kubernetes-integration | 8516364ebadda1ea82fdeb170527c23af9fea868 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/125/pull-gcp-filestore-csi-driver-kubernetes-integration/1405556308853133312/) | `/test pull-gcp-filestore-csi-driver-kubernetes-integration`
pull-gcp-filestore-csi-driver-e2e | 09c17c277cd847c6837c0fafdf38782ba52af833 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/125/pull-gcp-filestore-csi-driver-e2e/1405654631777832960/) | `/test pull-gcp-filestore-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=125). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<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. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
saikat-royc: /test pull-gcp-filestore-csi-driver-kubernetes-integration
saikat-royc: /test pull-gcp-filestore-csi-driver-kubernetes-integration
k8s-ci-robot: @saikat-royc: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-gcp-filestore-csi-driver-kubernetes-integration | 8516364ebadda1ea82fdeb170527c23af9fea868 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/125/pull-gcp-filestore-csi-driver-kubernetes-integration/1405556308853133312/) | `/test pull-gcp-filestore-csi-driver-kubernetes-integration`
pull-gcp-filestore-csi-driver-e2e | 38499d308c8a098e1839f1500116fb1546f5002c | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/125/pull-gcp-filestore-csi-driver-e2e/1405668894831546368/) | `/test pull-gcp-filestore-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=125). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<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. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
saikat-royc: /retest
k8s-ci-robot: @saikat-royc: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-gcp-filestore-csi-driver-kubernetes-integration | 38499d308c8a098e1839f1500116fb1546f5002c | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/125/pull-gcp-filestore-csi-driver-kubernetes-integration/1405668940566237184/) | `/test pull-gcp-filestore-csi-driver-kubernetes-integration`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=125). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<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. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @saikat-royc: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-gcp-filestore-csi-driver-kubernetes-integration | 38499d308c8a098e1839f1500116fb1546f5002c | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/125/pull-gcp-filestore-csi-driver-kubernetes-integration/1405668940566237184/) | `/test pull-gcp-filestore-csi-driver-kubernetes-integration`
pull-gcp-filestore-csi-driver-e2e | e3829564c2f6910f23679640289df9aa7a24fa6a | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/125/pull-gcp-filestore-csi-driver-e2e/1405749705190674432/) | `/test pull-gcp-filestore-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=125). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<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. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
saikat-royc: /retest
mattcary: /assign
k8s-ci-robot: @saikat-royc: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-gcp-filestore-csi-driver-kubernetes-integration | 38499d308c8a098e1839f1500116fb1546f5002c | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/125/pull-gcp-filestore-csi-driver-kubernetes-integration/1405668940566237184/) | `/test pull-gcp-filestore-csi-driver-kubernetes-integration`
pull-gcp-filestore-csi-driver-e2e | 23fcf445bb778281ada6997d4df84c1fe3f4ff83 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/125/pull-gcp-filestore-csi-driver-e2e/1405979983284801536/) | `/test pull-gcp-filestore-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=125). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<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. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
saikat-royc: /retest

Patch:
diff --git a/pkg/cloud_provider/file/fake.go b/pkg/cloud_provider/file/fake.go
index 82f70d73..00d92d7f 100644
--- a/pkg/cloud_provider/file/fake.go
+++ b/pkg/cloud_provider/file/fake.go
@@ -198,3 +198,33 @@ func notFoundError() *googleapi.Error {
 		},
 	}
 }
+
+type fakeBlockingServiceManager struct {
+	*fakeServiceManager
+	// 'OperationUnblocker' channel is used to block the execution of the respective function using it. This is done by sending a channel of empty struct over 'OperationUnblocker' channel, and wait until the tester gives a go-ahead to proceed further in the execution of the function.
+	OperationUnblocker chan chan struct{}
+}
+
+func NewFakeBlockingService(operationUnblocker chan chan struct{}) (Service, error) {
+	return &fakeBlockingServiceManager{
+		fakeServiceManager: &fakeServiceManager{
+			createdInstances: map[string]*ServiceInstance{},
+			backups:          map[string]*BackupInfo{},
+		},
+		OperationUnblocker: operationUnblocker,
+	}, nil
+}
+
+func (m *fakeBlockingServiceManager) CreateInstance(ctx context.Context, obj *ServiceInstance) (*ServiceInstance, error) {
+	execute := make(chan struct{})
+	m.OperationUnblocker <- execute
+	<-execute
+	return m.fakeServiceManager.CreateInstance(ctx, obj)
+}
+
+func (m *fakeBlockingServiceManager) DeleteInstance(ctx context.Context, obj *ServiceInstance) error {
+	execute := make(chan struct{})
+	m.OperationUnblocker <- execute
+	<-execute
+	return m.fakeServiceManager.DeleteInstance(ctx, obj)
+}
diff --git a/pkg/cloud_provider/file/file.go b/pkg/cloud_provider/file/file.go
index 7784ae7c..753cda17 100644
--- a/pkg/cloud_provider/file/file.go
+++ b/pkg/cloud_provider/file/file.go
@@ -147,7 +147,7 @@ func (manager *gcfsServiceManager) CreateInstance(ctx context.Context, obj *Serv
 	glog.V(4).Infof("For instance %v, waiting for create instance op %v to complete", obj.Name, op.Name)
 	err = manager.waitForOp(ctx, op)
 	if err != nil {
-		return nil, fmt.Errorf("WaitFor CreateInstance operation failed: %v", err)
+		return nil, fmt.Errorf("WaitFor CreateInstance op %s failed: %v", op.Name, err)
 	}
 	instance, err := manager.GetInstance(ctx, obj)
 	if err != nil {
@@ -192,9 +192,9 @@ func (manager *gcfsServiceManager) CreateInstanceFromBackupSource(ctx context.Co
 	}
 
 	glog.V(4).Infof("For instance %v, waiting for create instance op %v to complete", obj.Name, op.Name)
-	err = manager.waitForV1beta1Op(ctx, op)
+	err = manager.waitForOp(ctx, op)
 	if err != nil {
-		return nil, fmt.Errorf("WaitFor CreateInstance operation failed: %v", err)
+		return nil, fmt.Errorf("WaitFor CreateInstance op %s failed: %v", op.Name, err)
 	}
 	serviceInstance, err := manager.GetInstance(ctx, obj)
 	if err != nil {
@@ -264,35 +264,28 @@ func CompareInstances(a, b *ServiceInstance) error {
 }
 
 func (manager *gcfsServiceManager) DeleteInstance(ctx context.Context, obj *ServiceInstance) error {
-	instance, err := manager.GetInstance(ctx, obj)
-	if err != nil {
-		if IsNotFoundErr(err) {
-			glog.Infof("Instance %v not found", obj.Name)
-			return nil
-		}
-		return err
-	}
-
-	glog.Infof("Starting DeleteInstance cloud operation")
-	op, err := manager.instancesService.Delete(instanceURI(obj.Project, obj.Location, obj.Name)).Context(ctx).Do()
+	uri := instanceURI(obj.Project, obj.Location, obj.Name)
+	glog.V(4).Infof("Starting DeleteInstance cloud operation for instance %s", uri)
+	op, err := manager.instancesService.Delete(uri).Context(ctx).Do()
 	if err != nil {
 		return fmt.Errorf("DeleteInstance operation failed: %v", err)
 	}
 
+	glog.V(4).Infof("For instance %s, waiting for delete op %v to complete", uri, op.Name)
 	err = manager.waitForOp(ctx, op)
 	if err != nil {
-		return fmt.Errorf("WaitFor DeleteInstance operation failed: %v", err)
+		return fmt.Errorf("WaitFor DeleteInstance op %s failed: %v", op.Name, err)
 	}
 
-	instance, err = manager.GetInstance(ctx, obj)
+	instance, err := manager.GetInstance(ctx, obj)
 	if err != nil && !IsNotFoundErr(err) {
 		return fmt.Errorf("failed to get instance after deletion: %v", err)
 	}
 	if instance != nil {
-		return fmt.Errorf("instance %v still exists after delete operation", obj.Name)
+		return fmt.Errorf("instance %s still exists after delete operation in state %v", uri, instance.State)
 	}
 
-	glog.Infof("Instance %v has been deleted", obj.Name)
+	glog.Infof("Instance %s has been deleted", uri)
 	return nil
 }
 
@@ -347,7 +340,6 @@ func (manager *gcfsServiceManager) ResizeInstance(ctx context.Context, obj *Serv
 		},
 	}
 
-	glog.Infof("Starting Patch instance cloud operation for instance %s", instanceuri)
 	glog.V(4).Infof("Patching instance %v: location %v, tier %v, capacity %v, network %v, ipRange %v",
 		obj.Name,
 		obj.Location,
@@ -360,9 +352,10 @@ func (manager *gcfsServiceManager) ResizeInstance(ctx context.Context, obj *Serv
 		return nil, fmt.Errorf("Patch operation failed: %v", err)
 	}
 
+	glog.V(4).Infof("For instance %s, waiting for patch op %v to complete", instanceuri, op.Name)
 	err = manager.waitForOp(ctx, op)
 	if err != nil {
-		return nil, fmt.Errorf("WaitFor patch operation failed: %v", err)
+		return nil, fmt.Errorf("WaitFor patch op %s failed: %v", op.Name, err)
 	}
 
 	instance, err = manager.GetInstance(ctx, obj)
@@ -401,10 +394,10 @@ func (manager *gcfsServiceManager) CreateBackup(ctx context.Context, obj *Servic
 		return nil, fmt.Errorf("Create Backup operation failed: %v", err)
 	}
 
-	glog.V(4).Infof("Waiting for backup op %v to complete", opbackup.Name)
-	err = manager.waitForV1beta1Op(ctx, opbackup)
+	glog.V(4).Infof("For backup uri %s, waiting for backup op %v to complete", backupUri, opbackup.Name)
+	err = manager.waitForOp(ctx, opbackup)
 	if err != nil {
-		return nil, fmt.Errorf("WaitFor CreateBackup for source instance %v, backup Id: %v, operation failed: %v", backupSource, backupUri, err)
+		return nil, fmt.Errorf("WaitFor CreateBackup op %s for source instance %v, backup uri: %v, operation failed: %v", opbackup.Name, backupSource, backupUri, err)
 	}
 
 	backupObj, err := manager.backupService.Get(backupUri).Context(ctx).Do()
@@ -419,25 +412,15 @@ func (manager *gcfsServiceManager) CreateBackup(ctx context.Context, obj *Servic
 }
 
 func (manager *gcfsServiceManager) DeleteBackup(ctx context.Context, backupId string) error {
-	_, err := manager.backupService.Get(backupId).Context(ctx).Do()
-	if err != nil {
-		if IsNotFoundErr(err) {
-			glog.Infof("Backup %v not found", backupId)
-			return nil
-		}
-
-		return err
-	}
-
 	opbackup, err := manager.backupService.Delete(backupId).Context(ctx).Do()
 	if err != nil {
-		return fmt.Errorf("Delete backup operation failed: %v", err)
+		return fmt.Errorf("For backup Id %s, delete backup operation %s failed: %v", backupId, opbackup.Name, err)
 	}
 
-	glog.V(4).Infof("Waiting for backup op %v to complete", opbackup.Name)
-	err = manager.waitForV1beta1Op(ctx, opbackup)
+	glog.V(4).Infof("For backup Id %s, waiting for backup op %v to complete", backupId, opbackup.Name)
+	err = manager.waitForOp(ctx, opbackup)
 	if err != nil {
-		return fmt.Errorf("Delete backup: %v, operation failed: %v", backupId, err)
+		return fmt.Errorf("Delete backup: %v, op %s failed: %v", backupId, opbackup.Name, err)
 	}
 
 	glog.Infof("Backup %v successfully deleted", backupId)
@@ -505,26 +488,6 @@ func IsNotFoundErr(err error) bool {
 	return false
 }
 
-func (manager *gcfsServiceManager) waitForV1beta1Op(ctx context.Context, op *filev1.Operation) error {
-	return wait.Poll(5*time.Second, 5*time.Minute, func() (bool, error) {
-		pollOp, err := manager.operationsV1beta1Service.Get(op.Name).Context(ctx).Do()
-		if err != nil {
-			return false, err
-		}
-		return isV1beta1OpDone(pollOp)
-	})
-}
-
-func isV1beta1OpDone(op *filev1.Operation) (bool, error) {
-	if op == nil {
-		return false, nil
-	}
-	if op.Error != nil {
-		return true, fmt.Errorf("operation %v failed (%v): %v", op.Name, op.Error.Code, op.Error.Message)
-	}
-	return op.Done, nil
-}
-
 // This function returns the backup URI, the region that was picked to be the backup resource location and error.
 func CreateBackpURI(obj *ServiceInstance, backupName, backupLocation string) (string, string, error) {
 	region := backupLocation
diff --git a/pkg/csi_driver/controller.go b/pkg/csi_driver/controller.go
index b5a646a6..da5e5cf4 100644
--- a/pkg/csi_driver/controller.go
+++ b/pkg/csi_driver/controller.go
@@ -81,6 +81,7 @@ type controllerServerConfig struct {
 	fileService file.Service
 	cloud       *cloud.Cloud
 	ipAllocator *util.IPAllocator
+	volumeLocks *util.VolumeLocks
 }
 
 func newControllerServer(config *controllerServerConfig) csi.ControllerServer {
@@ -106,6 +107,17 @@ func (s *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolu
 	}
 	glog.V(5).Infof("Using capacity bytes %q for volume %q", capBytes, name)
 
+	newFiler, err := s.generateNewFileInstance(name, capBytes, req.GetParameters(), req.GetAccessibilityRequirements())
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+
+	volumeID := getVolumeIDFromFileInstance(newFiler, modeInstance)
+	if acquired := s.config.volumeLocks.TryAcquire(volumeID); !acquired {
+		return nil, status.Errorf(codes.Aborted, util.VolumeOperationAlreadyExistsFmt, volumeID)
+	}
+	defer s.config.volumeLocks.Release(volumeID)
+
 	sourceSnapshotId := ""
 	if req.GetVolumeContentSource() != nil {
 		if req.GetVolumeContentSource().GetVolume() != nil {
@@ -131,11 +143,6 @@ func (s *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolu
 		}
 	}
 
-	newFiler, err := s.generateNewFileInstance(name, capBytes, req.GetParameters(), req.GetAccessibilityRequirements())
-	if err != nil {
-		return nil, status.Error(codes.InvalidArgument, err.Error())
-	}
-
 	// Check if the instance already exists
 	filer, err := s.config.fileService.GetInstance(ctx, newFiler)
 	if err != nil && !file.IsNotFoundErr(err) {
@@ -148,7 +155,12 @@ func (s *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolu
 		if err = file.CompareInstances(newFiler, filer); err != nil {
 			return nil, status.Error(codes.AlreadyExists, err.Error())
 		}
-
+		// Check if the filestore instance is in the process of getting created.
+		if filer.State == "CREATING" {
+			msg := fmt.Sprintf("Volume %v not ready, current state: %v", name, filer.State)
+			glog.V(4).Infof(msg)
+			return nil, status.Error(codes.DeadlineExceeded, msg)
+		}
 		if filer.State != "READY" {
 			msg := fmt.Sprintf("Volume %v not ready, current state: %v", name, filer.State)
 			glog.V(4).Infof(msg)
@@ -187,6 +199,7 @@ func (s *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolu
 			filer, createErr = s.config.fileService.CreateInstance(ctx, newFiler)
 		}
 		if createErr != nil {
+			glog.Errorf("Create volume for volume Id %s failed: %v", volumeID, createErr)
 			return nil, status.Error(codes.Internal, createErr.Error())
 		}
 	}
@@ -229,6 +242,7 @@ func (s *controllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVolu
 	if volumeID == "" {
 		return nil, status.Error(codes.InvalidArgument, "volume id is empty")
 	}
+
 	filer, _, err := getFileInstanceFromID(volumeID)
 	if err != nil {
 		// An invalid ID should be treated as doesn't exist
@@ -236,9 +250,27 @@ func (s *controllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVolu
 		return &csi.DeleteVolumeResponse{}, nil
 	}
 
+	if acquired := s.config.volumeLocks.TryAcquire(volumeID); !acquired {
+		return nil, status.Errorf(codes.Aborted, util.VolumeOperationAlreadyExistsFmt, volumeID)
+	}
+	defer s.config.volumeLocks.Release(volumeID)
+
 	filer.Project = s.config.cloud.Project
+	filer, err = s.config.fileService.GetInstance(ctx, filer)
+	if err != nil {
+		if file.IsNotFoundErr(err) {
+			return &csi.DeleteVolumeResponse{}, nil
+		}
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	if filer.State == "DELETING" {
+		return nil, status.Errorf(codes.DeadlineExceeded, "Volume %s is in state: %s", volumeID, filer.State)
+	}
+
 	err = s.config.fileService.DeleteInstance(ctx, filer)
 	if err != nil {
+		glog.Errorf("Delete volume for volume Id %s failed: %v", volumeID, err)
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
@@ -408,6 +440,11 @@ func (s *controllerServer) ControllerExpandVolume(ctx context.Context, req *csi.
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
 
+	if acquired := s.config.volumeLocks.TryAcquire(volumeID); !acquired {
+		return nil, status.Errorf(codes.Aborted, util.VolumeOperationAlreadyExistsFmt, volumeID)
+	}
+	defer s.config.volumeLocks.Release(volumeID)
+
 	filer, _, err := getFileInstanceFromID(volumeID)
 	if err != nil {
 		glog.Errorf("failed to get instance for volumeID %v expansion, error: %v", volumeID, err)
@@ -541,6 +578,12 @@ func (s *controllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateSn
 	if len(volumeID) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "CreateSnapshot source volume ID must be provided")
 	}
+
+	if acquired := s.config.volumeLocks.TryAcquire(volumeID); !acquired {
+		return nil, status.Errorf(codes.Aborted, util.VolumeOperationAlreadyExistsFmt, volumeID)
+	}
+	defer s.config.volumeLocks.Release(volumeID)
+
 	filer, _, err := getFileInstanceFromID(volumeID)
 	if err != nil {
 		glog.Errorf("Failed to get instance for volumeID %v snapshot, error: %v", volumeID, err)
@@ -572,7 +615,10 @@ func (s *controllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateSn
 		if backupSourceCSIHandle != volumeID {
 			return nil, status.Error(codes.AlreadyExists, fmt.Sprintf("Backup already exists with a different source volume %s, input source volume %s", backupInfo.SourceVolumeHandle, volumeID))
 		}
-		// Check if backup is ready.
+		// Check if backup is in the process of getting created.
+		if backupInfo.Backup.State == "CREATING" || backupInfo.Backup.State == "FINALIZING" {
+			return nil, status.Error(codes.DeadlineExceeded, fmt.Sprintf("Backup %v not yet ready, current state %s", backupInfo.Backup.Name, backupInfo.Backup.State))
+		}
 		if backupInfo.Backup.State != "READY" {
 			return nil, status.Error(codes.Internal, fmt.Sprintf("Backup %v not yet ready, current state %s", backupInfo.Backup.Name, backupInfo.Backup.State))
 		}
@@ -594,6 +640,7 @@ func (s *controllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateSn
 
 	backupObj, err := s.config.fileService.CreateBackup(ctx, filer, req.Name, util.GetBackupLocation(req.GetParameters()))
 	if err != nil {
+		glog.Errorf("Create snapshot for volume Id %s failed: %v", volumeID, err)
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 	tp, err := util.ParseTimestamp(backupObj.CreateTime)
@@ -627,11 +674,25 @@ func (s *controllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteSn
 	}
 
 	if !isBackup {
-		glog.Errorf("Deletion of snapshot type %q not supported", id)
-		return nil, status.Error(codes.Internal, "Deletion of snapshot type not supported")
+		glog.Errorf("Deletion of volume snapshot type %q not supported", id)
+		return nil, status.Error(codes.Internal, "Deletion of volume snapshot type not supported")
+	}
+
+	backupInfo, err := s.config.fileService.GetBackup(ctx, id)
+	if err != nil {
+		if file.IsNotFoundErr(err) {
+			glog.Infof("Volume snapshot with ID %v not found", id)
+			return &csi.DeleteSnapshotResponse{}, nil
+		}
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	if backupInfo.Backup.State == "DELETING" {
+		return nil, status.Error(codes.DeadlineExceeded, fmt.Sprintf("Volume snapshot with ID %v is in state %s", id, backupInfo.Backup.State))
 	}
 
 	if err = s.config.fileService.DeleteBackup(ctx, id); err != nil {
+		glog.Errorf("Delete snapshot for backup Id %s failed: %v", id, err)
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
diff --git a/pkg/csi_driver/gcfs_driver.go b/pkg/csi_driver/gcfs_driver.go
index ca645ac8..7ea098c0 100644
--- a/pkg/csi_driver/gcfs_driver.go
+++ b/pkg/csi_driver/gcfs_driver.go
@@ -26,6 +26,7 @@ import (
 	"k8s.io/utils/mount"
 	cloud "sigs.k8s.io/gcp-filestore-csi-driver/pkg/cloud_provider"
 	metadataservice "sigs.k8s.io/gcp-filestore-csi-driver/pkg/cloud_provider/metadata"
+	"sigs.k8s.io/gcp-filestore-csi-driver/pkg/util"
 )
 
 type GCFSDriverConfig struct {
@@ -100,6 +101,7 @@ func NewGCFSDriver(config *GCFSDriverConfig) (*GCFSDriver, error) {
 			driver:      driver,
 			fileService: config.Cloud.File,
 			cloud:       config.Cloud,
+			volumeLocks: util.NewVolumeLocks(),
 		})
 	}
 
diff --git a/pkg/csi_driver/node.go b/pkg/csi_driver/node.go
index e9419a00..dc59eed5 100644
--- a/pkg/csi_driver/node.go
+++ b/pkg/csi_driver/node.go
@@ -29,6 +29,7 @@ import (
 	"google.golang.org/grpc/status"
 	"k8s.io/utils/mount"
 	"sigs.k8s.io/gcp-filestore-csi-driver/pkg/cloud_provider/metadata"
+	"sigs.k8s.io/gcp-filestore-csi-driver/pkg/util"
 )
 
 const (
@@ -46,6 +47,7 @@ type nodeServer struct {
 	driver      *GCFSDriver
 	mounter     mount.Interface
 	metaService metadata.Service
+	volumeLocks *util.VolumeLocks
 }
 
 func newNodeServer(driver *GCFSDriver, mounter mount.Interface, metaService metadata.Service) csi.NodeServer {
@@ -53,6 +55,7 @@ func newNodeServer(driver *GCFSDriver, mounter mount.Interface, metaService meta
 		driver:      driver,
 		mounter:     mounter,
 		metaService: metaService,
+		volumeLocks: util.NewVolumeLocks(),
 	}
 }
 
@@ -73,6 +76,12 @@ func (s *nodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublish
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
 
+	// Acquire a lock on the target path instead of volumeID, since we do not want to serialize multiple node publish calls on the same volume.
+	if acquired := s.volumeLocks.TryAcquire(targetPath); !acquired {
+		return nil, status.Errorf(codes.Aborted, util.VolumeOperationAlreadyExistsFmt, targetPath)
+	}
+	defer s.volumeLocks.Release(targetPath)
+
 	var err error
 	// FileSystem type
 	fstype := "nfs"
@@ -142,6 +151,12 @@ func (s *nodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpub
 		return nil, status.Error(codes.InvalidArgument, "NodeUnpublishVolume target path must be provided")
 	}
 
+	// Acquire a lock on the target path instead of volumeID, since we do not want to serialize multiple node unpublish calls on the same volume.
+	if acquired := s.volumeLocks.TryAcquire(targetPath); !acquired {
+		return nil, status.Errorf(codes.Aborted, util.VolumeOperationAlreadyExistsFmt, targetPath)
+	}
+	defer s.volumeLocks.Release(targetPath)
+
 	if err := mount.CleanupMountPoint(targetPath, s.mounter, false /* extensiveMountPointCheck */); err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
 	}
@@ -188,6 +203,12 @@ func (s *nodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolu
 	if err := validateVolumeAttributes(attr); err != nil {
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
+
+	if acquired := s.volumeLocks.TryAcquire(volumeID); !acquired {
+		return nil, status.Errorf(codes.Aborted, util.VolumeOperationAlreadyExistsFmt, volumeID)
+	}
+	defer s.volumeLocks.Release(volumeID)
+
 	// Mount source
 	source := fmt.Sprintf("%s:/%s", attr[attrIP], attr[attrVolume])
 	mounted, err := s.isDirMounted(stagingTargetPath)
@@ -245,6 +266,11 @@ func (s *nodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstage
 		return nil, status.Error(codes.InvalidArgument, "NodeUnstageVolume Staging Target Path must be provided")
 	}
 
+	if acquired := s.volumeLocks.TryAcquire(volumeID); !acquired {
+		return nil, status.Errorf(codes.Aborted, util.VolumeOperationAlreadyExistsFmt, volumeID)
+	}
+	defer s.volumeLocks.Release(volumeID)
+
 	if err := mount.CleanupMountPoint(stagingTargetPath, s.mounter, false /* extensiveMountPointCheck */); err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
 	}
diff --git a/pkg/util/volume_lock.go b/pkg/util/volume_lock.go
new file mode 100644
index 00000000..bec5a34f
--- /dev/null
+++ b/pkg/util/volume_lock.go
@@ -0,0 +1,58 @@
+/*
+Copyright 2021 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License 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 util
+
+import (
+	"sync"
+
+	"k8s.io/apimachinery/pkg/util/sets"
+)
+
+const (
+	VolumeOperationAlreadyExistsFmt = "An operation with the given volume key %s already exists"
+)
+
+// VolumeLocks implements a map with atomic operations. It stores a set of all volume IDs
+// with an ongoing operation.
+type VolumeLocks struct {
+	locks sets.String
+	mux   sync.Mutex
+}
+
+func NewVolumeLocks() *VolumeLocks {
+	return &VolumeLocks{
+		locks: sets.NewString(),
+	}
+}
+
+// TryAcquire tries to acquire the lock for operating on volumeID and returns true if successful.
+// If another operation is already using volumeID, returns false.
+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/csi_driver/controller_test.go b/pkg/csi_driver/controller_test.go
index 8f0f740d..5e5a6262 100644
--- a/pkg/csi_driver/controller_test.go
+++ b/pkg/csi_driver/controller_test.go
@@ -22,6 +22,8 @@ import (
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	"golang.org/x/net/context"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"k8s.io/apimachinery/pkg/util/sets"
 	cloud "sigs.k8s.io/gcp-filestore-csi-driver/pkg/cloud_provider"
 	"sigs.k8s.io/gcp-filestore-csi-driver/pkg/cloud_provider/file"
@@ -33,6 +35,7 @@ const (
 	testLocation         = "us-central1-c"
 	testIP               = "1.1.1.1"
 	testCSIVolume        = "test-csi"
+	testCSIVolume2       = "test-csi-2"
 	testVolumeID         = "modeInstance/us-central1-c/test-csi/vol1"
 	testReservedIPV4CIDR = "192.168.92.0/26"
 	testBytes            = 1 * util.Tb
@@ -52,6 +55,25 @@ func initTestController(t *testing.T) csi.ControllerServer {
 		driver:      initTestDriver(t),
 		fileService: fileService,
 		cloud:       cloudProvider,
+		volumeLocks: util.NewVolumeLocks(),
+	})
+}
+
+func initBlockingTestController(t *testing.T, operationUnblocker chan chan struct{}) csi.ControllerServer {
+	fileService, err := file.NewFakeBlockingService(operationUnblocker)
+	if err != nil {
+		t.Fatalf("failed to initialize blocking GCFS service: %v", err)
+	}
+
+	cloudProvider, err := cloud.NewFakeCloud()
+	if err != nil {
+		t.Fatalf("Failed to get cloud provider: %v", err)
+	}
+	return newControllerServer(&controllerServerConfig{
+		driver:      initTestDriver(t),
+		fileService: fileService,
+		cloud:       cloudProvider,
+		volumeLocks: util.NewVolumeLocks(),
 	})
 }
 
@@ -643,3 +665,134 @@ func TestGetZonesFromTopology(t *testing.T) {
 		})
 	}
 }
+
+type RequestConfig struct {
+	CreateVolReq  *csi.CreateVolumeRequest
+	DeleteVolReq  *csi.DeleteVolumeRequest
+	CreateSnapReq *csi.CreateSnapshotRequest
+	DeleteSnapReq *csi.DeleteSnapshotRequest
+	ExpandVolReq  *csi.ControllerExpandVolumeRequest
+}
+
+func TestVolumeOperationLocks(t *testing.T) {
+	// A channel of size 1 is sufficient, because the caller of runRequest() in below steps immediately blocks and retrieves the channel of empty struct from 'operationUnblocker' channel. The test steps are such that, atmost one function pushes items on the 'operationUnblocker' channel, to indicate that the function is blocked and waiting for a signal to proceed futher in the execution.
+	operationUnblocker := make(chan chan struct{}, 1)
+	cs := initBlockingTestController(t, operationUnblocker)
+	runRequest := func(req *RequestConfig) <-chan error {
+		resp := make(chan error)
+		go func() {
+			var err error
+			if req.CreateVolReq != nil {
+				_, err = cs.CreateVolume(context.Background(), req.CreateVolReq)
+			} else if req.DeleteVolReq != nil {
+				_, err = cs.DeleteVolume(context.Background(), req.DeleteVolReq)
+			} else if req.CreateSnapReq != nil {
+				_, err = cs.CreateSnapshot(context.Background(), req.CreateSnapReq)
+			} else if req.DeleteSnapReq != nil {
+				_, err = cs.DeleteSnapshot(context.Background(), req.DeleteSnapReq)
+			} else if req.ExpandVolReq != nil {
+				_, err = cs.ControllerExpandVolume(context.Background(), req.ExpandVolReq)
+			}
+			resp <- err
+		}()
+		return resp
+	}
+
+	req := &csi.CreateVolumeRequest{
+		Name: testCSIVolume,
+		VolumeCapabilities: []*csi.VolumeCapability{
+			{
+				AccessType: &csi.VolumeCapability_Mount{
+					Mount: &csi.VolumeCapability_MountVolume{},
+				},
+				AccessMode: &csi.VolumeCapability_AccessMode{
+					Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+				},
+			},
+		},
+	}
+	// Block first CreateVolume request after it has acquired the lock.
+	resp := runRequest(&RequestConfig{CreateVolReq: req})
+	createOpUnblocker := <-operationUnblocker
+
+	// Second CreateVolume request on the same volume should fail to acquire lock and return Aborted error.
+	createResp2 := runRequest(&RequestConfig{CreateVolReq: req})
+	ValidateExpectedError(t, createResp2, operationUnblocker, codes.Aborted)
+
+	// Delete Volume request on the same volume should fail to acquire lock and return Aborted error.
+	delResp := runRequest(&RequestConfig{DeleteVolReq: &csi.DeleteVolumeRequest{
+		VolumeId: testVolumeID,
+	}})
+	ValidateExpectedError(t, delResp, operationUnblocker, codes.Aborted)
+
+	// Create a snapshot on the same volume should fail to acquire lock and return Aborted error.
+	createSnapResp := runRequest(&RequestConfig{
+		CreateSnapReq: &csi.CreateSnapshotRequest{
+			Name:           "test-snap",
+			SourceVolumeId: testVolumeID,
+		},
+	})
+	ValidateExpectedError(t, createSnapResp, operationUnblocker, codes.Aborted)
+
+	// ControllerExapnd request on the same volume should fail to acquire lock and return Aborted error.
+	expandVolResp := runRequest(&RequestConfig{
+		ExpandVolReq: &csi.ControllerExpandVolumeRequest{
+			VolumeId: testVolumeID,
+		},
+	})
+	ValidateExpectedError(t, expandVolResp, operationUnblocker, codes.Aborted)
+
+	// Send a create volume request for a different volume. This is expected to succeed.
+	vol2CreateVolResp := runRequest(&RequestConfig{CreateVolReq: &csi.CreateVolumeRequest{
+		Name: testCSIVolume2,
+		VolumeCapabilities: []*csi.VolumeCapability{
+			{
+				AccessType: &csi.VolumeCapability_Mount{
+					Mount: &csi.VolumeCapability_MountVolume{},
+				},
+				AccessMode: &csi.VolumeCapability_AccessMode{
+					Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+				},
+			},
+		},
+	}})
+	execVol2CreateVol := <-operationUnblocker
+	execVol2CreateVol <- struct{}{}
+	if err := <-vol2CreateVolResp; err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+
+	// Unblock first CreateVolume request and let it run to completion.
+	createOpUnblocker <- struct{}{}
+	if err := <-resp; err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+	// Delete the first volume, no error expected.
+	delResp = runRequest(&RequestConfig{DeleteVolReq: &csi.DeleteVolumeRequest{
+		VolumeId: testVolumeID,
+	}})
+	execDelVol := <-operationUnblocker
+	execDelVol <- struct{}{}
+	if err := <-delResp; err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+}
+
+func ValidateExpectedError(t *testing.T, errResp <-chan error, operationUnblocker chan chan struct{}, expectedErrorCode codes.Code) {
+	select {
+	case err := <-errResp:
+		if err != nil {
+			serverError, ok := status.FromError(err)
+			if !ok {
+				t.Fatalf("Could not get error status code from err: %v", err)
+			}
+			if serverError.Code() != codes.Aborted {
+				t.Errorf("Expected error code: %v, got: %v. err : %v", codes.Aborted, serverError.Code(), err)
+			}
+		} else {
+			t.Errorf("Expected error: %v, got no error", codes.Aborted)
+		}
+	case <-operationUnblocker:
+		t.Errorf("The operation should have been aborted, but was started")
+	}
+}
diff --git a/pkg/csi_driver/node_test.go b/pkg/csi_driver/node_test.go
index c350de48..1dbba1af 100644
--- a/pkg/csi_driver/node_test.go
+++ b/pkg/csi_driver/node_test.go
@@ -24,6 +24,7 @@ import (
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	"golang.org/x/net/context"
+	"google.golang.org/grpc/codes"
 	"k8s.io/utils/mount"
 	"sigs.k8s.io/gcp-filestore-csi-driver/pkg/cloud_provider/metadata"
 )
@@ -494,3 +495,176 @@ func validateMountPoint(t *testing.T, name string, fm *mount.FakeMounter, e *mou
 		}
 	}
 }
+
+type FakeBlockingMounter struct {
+	*mount.FakeMounter
+	// 'OperationUnblocker' channel is used to block the execution of the respective function using it. This is done by sending a channel of empty struct over 'OperationUnblocker' channel, and wait until the tester gives a go-ahead to proceed further in the execution of the function.
+	OperationUnblocker chan chan struct{}
+}
+
+func NewFakeBlockingMounter(operationUnblocker chan chan struct{}) *FakeBlockingMounter {
+	return &FakeBlockingMounter{
+		FakeMounter:        &mount.FakeMounter{MountPoints: []mount.MountPoint{}},
+		OperationUnblocker: operationUnblocker,
+	}
+}
+
+func (m *FakeBlockingMounter) Mount(source string, target string, fstype string, options []string) error {
+	execute := make(chan struct{})
+	m.OperationUnblocker <- execute
+	<-execute
+	return m.FakeMounter.Mount(source, target, fstype, options)
+}
+
+func initBlockingTestNodeServer(t *testing.T, operationUnblocker chan chan struct{}) *nodeServerTestEnv {
+	mounter := NewFakeBlockingMounter(operationUnblocker)
+	metaserice, err := metadata.NewFakeService()
+	if err != nil {
+		t.Fatalf("Failed to init metadata service")
+	}
+	return &nodeServerTestEnv{
+		ns: newNodeServer(initTestDriver(t), mounter, metaserice),
+		fm: nil,
+	}
+}
+
+type NodeRequestConfig struct {
+	NodePublishReq   *csi.NodePublishVolumeRequest
+	NodeUnpublishReq *csi.NodeUnpublishVolumeRequest
+	NodeStageReq     *csi.NodeStageVolumeRequest
+	NodeUnstageReq   *csi.NodeUnstageVolumeRequest
+}
+
+func TestConcurrentMounts(t *testing.T) {
+	// A channel of size 1 is sufficient, because the caller of runRequest() in below steps immediately blocks and retrieves the channel of empty struct from 'operationUnblocker' channel. The test steps are such that, atmost one function pushes items on the 'operationUnblocker' channel, to indicate that the function is blocked and waiting for a signal to proceed futher in the execution.
+	operationUnblocker := make(chan chan struct{}, 1)
+	ns := initBlockingTestNodeServer(t, operationUnblocker)
+	basePath, err := ioutil.TempDir("", "node-publish-")
+	if err != nil {
+		t.Fatalf("failed to setup testdir: %v", err)
+	}
+	stagingTargetPath := filepath.Join(basePath, "staging")
+	targetPath1 := filepath.Join(basePath, "target1")
+	targetPath2 := filepath.Join(basePath, "target2")
+
+	runRequest := func(req *NodeRequestConfig) <-chan error {
+		resp := make(chan error)
+		go func() {
+			var err error
+			if req.NodePublishReq != nil {
+				_, err = ns.ns.NodePublishVolume(context.Background(), req.NodePublishReq)
+			} else if req.NodeUnpublishReq != nil {
+				_, err = ns.ns.NodeUnpublishVolume(context.Background(), req.NodeUnpublishReq)
+			} else if req.NodeStageReq != nil {
+				_, err = ns.ns.NodeStageVolume(context.Background(), req.NodeStageReq)
+			} else if req.NodeUnstageReq != nil {
+				_, err = ns.ns.NodeUnstageVolume(context.Background(), req.NodeUnstageReq)
+			}
+			resp <- err
+		}()
+		return resp
+	}
+
+	// Node stage blocked after lock acquire.
+	resp := runRequest(&NodeRequestConfig{
+		NodeStageReq: &csi.NodeStageVolumeRequest{
+			VolumeId:          testVolumeID,
+			StagingTargetPath: stagingTargetPath,
+			VolumeCapability:  testVolumeCapability,
+			VolumeContext:     testVolumeAttributes,
+		},
+	})
+	nodestageOpUnblocker := <-operationUnblocker
+
+	// Same volume ID node stage should fail to acquire lock and return Aborted error.
+	stageResp2 := runRequest(&NodeRequestConfig{
+		NodeStageReq: &csi.NodeStageVolumeRequest{
+			VolumeId:          testVolumeID,
+			StagingTargetPath: stagingTargetPath,
+			VolumeCapability:  testVolumeCapability,
+			VolumeContext:     testVolumeAttributes,
+		},
+	})
+	ValidateExpectedError(t, stageResp2, operationUnblocker, codes.Aborted)
+
+	// Same volume ID node unstage should fail to acquire lock and return Aborted error.
+	unstageResp := runRequest(&NodeRequestConfig{
+		NodeUnstageReq: &csi.NodeUnstageVolumeRequest{
+			VolumeId:          testVolumeID,
+			StagingTargetPath: stagingTargetPath,
+		},
+	})
+	ValidateExpectedError(t, unstageResp, operationUnblocker, codes.Aborted)
+
+	// Unblock first node stage. Success expected.
+	nodestageOpUnblocker <- struct{}{}
+	if err := <-resp; err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+
+	// Node publish blocked after lock acquire on the 'targetPath1'.
+	targetPath1Publishresp := runRequest(&NodeRequestConfig{
+		NodePublishReq: &csi.NodePublishVolumeRequest{
+			VolumeId:          testVolumeID,
+			StagingTargetPath: stagingTargetPath,
+			TargetPath:        targetPath1,
+			VolumeCapability:  testVolumeCapability,
+			VolumeContext:     testVolumeAttributes,
+		},
+	})
+	nodepublishOpTargetPath1Unblocker := <-operationUnblocker
+
+	// Node publish for the same target path should fail to acquire lock and return Aborted error.
+	targetPath1Publishresp2 := runRequest(&NodeRequestConfig{
+		NodePublishReq: &csi.NodePublishVolumeRequest{
+			VolumeId:          testVolumeID,
+			StagingTargetPath: stagingTargetPath,
+			TargetPath:        targetPath1,
+			VolumeCapability:  testVolumeCapability,
+			VolumeContext:     testVolumeAttributes,
+		},
+	})
+	ValidateExpectedError(t, targetPath1Publishresp2, operationUnblocker, codes.Aborted)
+
+	// Node unpublish for the same target path should fail to acquire lock and return Aborted error.
+	targetPath1Unpublishresp := runRequest(&NodeRequestConfig{
+		NodeUnpublishReq: &csi.NodeUnpublishVolumeRequest{
+			VolumeId:   testVolumeID,
+			TargetPath: targetPath1,
+		},
+	})
+	ValidateExpectedError(t, targetPath1Unpublishresp, operationUnblocker, codes.Aborted)
+
+	// Node publish succeeds for a second target path 'targetPath2'.
+	targetPath2Publishresp2 := runRequest(&NodeRequestConfig{
+		NodePublishReq: &csi.NodePublishVolumeRequest{
+			VolumeId:          testVolumeID,
+			StagingTargetPath: stagingTargetPath,
+			TargetPath:        targetPath2,
+			VolumeCapability:  testVolumeCapability,
+			VolumeContext:     testVolumeAttributes,
+		},
+	})
+	nodepublishOpTargetPath2Unblocker := <-operationUnblocker
+	nodepublishOpTargetPath2Unblocker <- struct{}{}
+	if err := <-targetPath2Publishresp2; err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+
+	// Node unpublish succeeds for second target path.
+	targetPath2Unpublishresp := runRequest(&NodeRequestConfig{
+		NodeUnpublishReq: &csi.NodeUnpublishVolumeRequest{
+			VolumeId:   testVolumeID,
+			TargetPath: targetPath2,
+		},
+	})
+	if err := <-targetPath2Unpublishresp; err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+
+	// Unblock first node publish, and success expected.
+	nodepublishOpTargetPath1Unblocker <- struct{}{}
+	if err := <-targetPath1Publishresp; err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+}