Repository: kubernetes-sigs/gcp-compute-persistent-disk-csi-driver
Pull request #1733 (2024-06-04 05:08:01)

Add support for "multi-zone" volume handle provisioning and deletion
**What type of PR is this?**
/kind feature

**What this PR does / why we need it**:
Add support for `multi-zone` volume handle provisioning. Volumes created with the `enable-multi-zone-provisioning: true` parameter will be created with multiple underlying disks, one per specified zone. The underlying volume returned will specify the keyword `multi-zone` in place of the `zone` field. See https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/pull/1616 for more details on the syntax.
 1) If there are zones specified in the requisite topologies, one disk will be created per zone
 2) If there are no zones specified in the requisite topologies, one disk will be created per zone in all zones in the driver's default region (where the particular disk type is supported).

Multi-zone volumes can be created from three sources:
1) PD images (ROX mode)
2) PD Snapshots (ROX mode)
3) Empty disks (RWO mode)

Deletion will inspect all disks that match the volume handle name, for all zones in the driver's default region.

**Special notes for your reviewer**:

This adds some additional e2e tests, that are not yet runnable, due to API support being in alpha. These have been run on a locally project that has alpha support, and can be replicated by running:

```
export RUN_MULTI_ZONE_TESTS=true
ginkgo --v "test/e2e/tests" -- --project "${PROJECT}" --service-account "${IAM_NAME}" --v=6 --logtostderr --ginkgo.focus "'multi-zone'" --extra-driver-flags=--set-access-mode-gce-api-version=alpha
```

**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
Add support for provisioning multi-zone volume handles
```


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

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/pull/1733#" title="Author self-approved">pwschuurman</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fgcp-compute-persistent-disk-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-compute-persistent-disk-csi-driver/blob/master/OWNERS)~~ [pwschuurman]

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: @pwschuurman: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-verify | fdab3e2f5c745783906ce86ec5c47c4e1b16b816 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_gcp-compute-persistent-disk-csi-driver/1733/pull-gcp-compute-persistent-disk-csi-driver-verify/1797857825683148800) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-verify`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-compute-persistent-disk-csi-driver&pr=1733). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Apwschuurman). 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-compute-persistent-disk-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-sigs/prow](https://github.com/kubernetes-sigs/prow/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: @pwschuurman: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-verify | fdab3e2f5c745783906ce86ec5c47c4e1b16b816 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_gcp-compute-persistent-disk-csi-driver/1733/pull-gcp-compute-persistent-disk-csi-driver-verify/1797857825683148800) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-verify`
pull-gcp-compute-persistent-disk-csi-driver-e2e | fdab3e2f5c745783906ce86ec5c47c4e1b16b816 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_gcp-compute-persistent-disk-csi-driver/1733/pull-gcp-compute-persistent-disk-csi-driver-e2e/1797857825465044992) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-compute-persistent-disk-csi-driver&pr=1733). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Apwschuurman). 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-compute-persistent-disk-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-sigs/prow](https://github.com/kubernetes-sigs/prow/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: @pwschuurman: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-e2e | fdab3e2f5c745783906ce86ec5c47c4e1b16b816 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_gcp-compute-persistent-disk-csi-driver/1733/pull-gcp-compute-persistent-disk-csi-driver-e2e/1797857825465044992) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-e2e`
pull-gcp-compute-persistent-disk-csi-driver-verify | 3ed2e0a7498ebccb8bdc3d96655f53bf94ccfef4 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_gcp-compute-persistent-disk-csi-driver/1733/pull-gcp-compute-persistent-disk-csi-driver-verify/1798116242138075136) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-verify`
pull-gcp-compute-persistent-disk-csi-driver-unit | 3ed2e0a7498ebccb8bdc3d96655f53bf94ccfef4 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_gcp-compute-persistent-disk-csi-driver/1733/pull-gcp-compute-persistent-disk-csi-driver-unit/1798116242079354880) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-unit`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-compute-persistent-disk-csi-driver&pr=1733). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Apwschuurman). 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-compute-persistent-disk-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-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
mattcary: /assign @mattcary 
k8s-ci-robot: @pwschuurman: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-e2e | 95ebafa66e82644bbc6bb43d7ed974b90a293580 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_gcp-compute-persistent-disk-csi-driver/1733/pull-gcp-compute-persistent-disk-csi-driver-e2e/1798580345050763264) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-compute-persistent-disk-csi-driver&pr=1733). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Apwschuurman). 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-compute-persistent-disk-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-sigs/prow](https://github.com/kubernetes-sigs/prow/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: @pwschuurman: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-e2e | 95ebafa66e82644bbc6bb43d7ed974b90a293580 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_gcp-compute-persistent-disk-csi-driver/1733/pull-gcp-compute-persistent-disk-csi-driver-e2e/1798770468023439360) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-compute-persistent-disk-csi-driver&pr=1733). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Apwschuurman). 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-compute-persistent-disk-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-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
pwschuurman: /retest
k8s-ci-robot: @pwschuurman: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-e2e | 333a6e54d628057acb7f235d328bc839371b9dcf | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_gcp-compute-persistent-disk-csi-driver/1733/pull-gcp-compute-persistent-disk-csi-driver-e2e/1799211382936178688) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-compute-persistent-disk-csi-driver&pr=1733). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Apwschuurman). 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-compute-persistent-disk-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-sigs/prow](https://github.com/kubernetes-sigs/prow/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: @pwschuurman: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-unit | 45c46249326d5e3dde37d8830e7f73d08145edbc | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_gcp-compute-persistent-disk-csi-driver/1733/pull-gcp-compute-persistent-disk-csi-driver-unit/1800942075110232064) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-unit`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-compute-persistent-disk-csi-driver&pr=1733). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Apwschuurman). 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-compute-persistent-disk-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-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->

Patch:
diff --git a/Dockerfile b/Dockerfile
index e6641def..e40187e5 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM --platform=$BUILDPLATFORM golang:1.22.4 as builder
+FROM --platform=$BUILDPLATFORM golang:1.22.3 as builder
 
 ARG STAGINGVERSION
 ARG TARGETPLATFORM
diff --git a/Dockerfile.Windows b/Dockerfile.Windows
index bc2570b0..f5b0fee9 100644
--- a/Dockerfile.Windows
+++ b/Dockerfile.Windows
@@ -13,7 +13,7 @@
 # limitations under the License.
 
 ARG BASE_IMAGE
-FROM --platform=$BUILDPLATFORM golang:1.22.4 AS builder
+FROM --platform=$BUILDPLATFORM golang:1.22.3 AS builder
 
 ARG TARGETPLATFORM
 ARG STAGINGVERSION
diff --git a/Dockerfile.debug b/Dockerfile.debug
index dc8a2c13..11e8701a 100644
--- a/Dockerfile.debug
+++ b/Dockerfile.debug
@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM golang:1.22.4 as builder
+FROM golang:1.22.3 as builder
 WORKDIR /go/src/sigs.k8s.io/gcp-compute-persistent-disk-csi-driver
 ADD . .
 
diff --git a/go.mod b/go.mod
index 6a875324..0430ed2a 100644
--- a/go.mod
+++ b/go.mod
@@ -22,11 +22,11 @@ require (
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.21.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.21.0
 	go.opentelemetry.io/otel/sdk v1.24.0
-	golang.org/x/oauth2 v0.21.0
+	golang.org/x/oauth2 v0.20.0
 	golang.org/x/sys v0.20.0
 	golang.org/x/time v0.5.0
-	google.golang.org/api v0.183.0
-	google.golang.org/genproto v0.0.0-20240528184218-531527333157
+	google.golang.org/api v0.182.0
+	google.golang.org/genproto v0.0.0-20240401170217-c3f982113cda
 	google.golang.org/grpc v1.64.0
 	google.golang.org/protobuf v1.34.1
 	gopkg.in/gcfg.v1 v1.2.3
@@ -43,7 +43,7 @@ require (
 
 require (
 	cloud.google.com/go v0.114.0 // indirect
-	cloud.google.com/go/auth v0.5.1 // indirect
+	cloud.google.com/go/auth v0.5.0 // indirect
 	cloud.google.com/go/auth/oauth2adapt v0.2.2 // indirect
 	cloud.google.com/go/iam v1.1.8 // indirect
 	cloud.google.com/go/longrunning v0.5.7 // indirect
@@ -107,8 +107,8 @@ require (
 	golang.org/x/term v0.20.0 // indirect
 	golang.org/x/text v0.15.0 // indirect
 	golang.org/x/tools v0.21.0 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20240521202816-d264139d666e // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20240528184218-531527333157 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20240513163218-0867130af1f8 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240521202816-d264139d666e // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect
 	gopkg.in/warnings.v0 v0.1.2 // indirect
diff --git a/go.sum b/go.sum
index bbe7ccd2..841ea000 100644
--- a/go.sum
+++ b/go.sum
@@ -110,8 +110,8 @@ cloud.google.com/go/assuredworkloads v1.7.0/go.mod h1:z/736/oNmtGAyU47reJgGN+KVo
 cloud.google.com/go/assuredworkloads v1.8.0/go.mod h1:AsX2cqyNCOvEQC8RMPnoc0yEarXQk6WEKkxYfL6kGIo=
 cloud.google.com/go/assuredworkloads v1.9.0/go.mod h1:kFuI1P78bplYtT77Tb1hi0FMxM0vVpRC7VVoJC3ZoT0=
 cloud.google.com/go/assuredworkloads v1.10.0/go.mod h1:kwdUQuXcedVdsIaKgKTp9t0UJkE5+PAVNhdQm4ZVq2E=
-cloud.google.com/go/auth v0.5.1 h1:0QNO7VThG54LUzKiQxv8C6x1YX7lUrzlAa1nVLF8CIw=
-cloud.google.com/go/auth v0.5.1/go.mod h1:vbZT8GjzDf3AVqCcQmqeeM32U9HBFc32vVVAbwDsa6s=
+cloud.google.com/go/auth v0.5.0 h1:GtSZfKJkPrZi/s3AkiHnUYVI4dTP/kg8+I3unm0omag=
+cloud.google.com/go/auth v0.5.0/go.mod h1:Kqvlz1cf1sNA0D+sYJnkPQOP+JMHkuHeIgVmCRtZOLc=
 cloud.google.com/go/auth/oauth2adapt v0.2.2 h1:+TTV8aXpjeChS9M+aTtN/TjdQnzJvmzKFt//oWu7HX4=
 cloud.google.com/go/auth/oauth2adapt v0.2.2/go.mod h1:wcYjgpZI9+Yu7LyYBg4pqSiaRkfEK3GQcpb7C/uyF1Q=
 cloud.google.com/go/automl v1.5.0/go.mod h1:34EjfoFGMZ5sgJ9EoLsRtdPSNZLcfflJR39VbVNS2M0=
@@ -2491,8 +2491,8 @@ golang.org/x/oauth2 v0.0.0-20221014153046-6fdb5e3db783/go.mod h1:h4gKUeWbJ4rQPri
 golang.org/x/oauth2 v0.4.0/go.mod h1:RznEsdpjGAINPTOF0UH/t+xJ75L18YO3Ho6Pyn+uRec=
 golang.org/x/oauth2 v0.5.0/go.mod h1:9/XBHVqLaWO3/BRHs5jbpYCnOZVjj5V0ndyaAM7KB4I=
 golang.org/x/oauth2 v0.6.0/go.mod h1:ycmewcwgD4Rpr3eZJLSB4Kyyljb3qDh40vJ8STE5HKw=
-golang.org/x/oauth2 v0.21.0 h1:tsimM75w1tF/uws5rbeHzIWxEqElMehnc+iW793zsZs=
-golang.org/x/oauth2 v0.21.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
+golang.org/x/oauth2 v0.20.0 h1:4mQdhULixXKP1rwYBW0vAijoXnkTG0BLCDRzfe1idMo=
+golang.org/x/oauth2 v0.20.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -2902,8 +2902,8 @@ google.golang.org/api v0.108.0/go.mod h1:2Ts0XTHNVWxypznxWOYUeI4g3WdP9Pk2Qk58+a/
 google.golang.org/api v0.110.0/go.mod h1:7FC4Vvx1Mooxh8C5HWjzZHcavuS2f6pmJpZx60ca7iI=
 google.golang.org/api v0.111.0/go.mod h1:qtFHvU9mhgTJegR31csQ+rwxyUTHOKFqCKWp1J0fdw0=
 google.golang.org/api v0.114.0/go.mod h1:ifYI2ZsFK6/uGddGfAD5BMxlnkBqCmqHSDUVi45N5Yg=
-google.golang.org/api v0.183.0 h1:PNMeRDwo1pJdgNcFQ9GstuLe/noWKIc89pRWRLMvLwE=
-google.golang.org/api v0.183.0/go.mod h1:q43adC5/pHoSZTx5h2mSmdF7NcyfW9JuDyIOJAgS9ZQ=
+google.golang.org/api v0.182.0 h1:if5fPvudRQ78GeRx3RayIoiuV7modtErPIZC/T2bIvE=
+google.golang.org/api v0.182.0/go.mod h1:cGhjy4caqA5yXRzEhkHI8Y9mfyC2VLTlER2l08xaqtM=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
 google.golang.org/appengine v1.2.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
 google.golang.org/appengine v1.3.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
@@ -3075,16 +3075,16 @@ google.golang.org/genproto v0.0.0-20230330154414-c0448cd141ea/go.mod h1:UUQDJDOl
 google.golang.org/genproto v0.0.0-20230331144136-dcfb400f0633/go.mod h1:UUQDJDOlWu4KYeJZffbWgBkS1YFobzKbLVfK69pe0Ak=
 google.golang.org/genproto v0.0.0-20230525234025-438c736192d0/go.mod h1:9ExIQyXL5hZrHzQceCwuSYwZZ5QZBazOcprJ5rgs3lY=
 google.golang.org/genproto v0.0.0-20230526161137-0005af68ea54/go.mod h1:zqTuNwFlFRsw5zIts5VnzLQxSRqh+CGOTVMlYbY0Eyk=
-google.golang.org/genproto v0.0.0-20240528184218-531527333157 h1:u7WMYrIrVvs0TF5yaKwKNbcJyySYf+HAIFXxWltJOXE=
-google.golang.org/genproto v0.0.0-20240528184218-531527333157/go.mod h1:ubQlAQnzejB8uZzszhrTCU2Fyp6Vi7ZE5nn0c3W8+qQ=
+google.golang.org/genproto v0.0.0-20240401170217-c3f982113cda h1:wu/KJm9KJwpfHWhkkZGohVC6KRrc1oJNr4jwtQMOQXw=
+google.golang.org/genproto v0.0.0-20240401170217-c3f982113cda/go.mod h1:g2LLCvCeCSir/JJSWosk19BR4NVxGqHUC6rxIRsd7Aw=
 google.golang.org/genproto/googleapis/api v0.0.0-20230525234020-1aefcd67740a/go.mod h1:ts19tUU+Z0ZShN1y3aPyq2+O3d5FUNNgT6FtOzmrNn8=
 google.golang.org/genproto/googleapis/api v0.0.0-20230525234035-dd9d682886f9/go.mod h1:vHYtlOoi6TsQ3Uk2yxR7NI5z8uoV+3pZtR4jmHIkRig=
-google.golang.org/genproto/googleapis/api v0.0.0-20240521202816-d264139d666e h1:SkdGTrROJl2jRGT/Fxv5QUf9jtdKCQh4KQJXbXVLAi0=
-google.golang.org/genproto/googleapis/api v0.0.0-20240521202816-d264139d666e/go.mod h1:LweJcLbyVij6rCex8YunD8DYR5VDonap/jYl3ZRxcIU=
+google.golang.org/genproto/googleapis/api v0.0.0-20240513163218-0867130af1f8 h1:W5Xj/70xIA4x60O/IFyXivR5MGqblAb8R3w26pnD6No=
+google.golang.org/genproto/googleapis/api v0.0.0-20240513163218-0867130af1f8/go.mod h1:vPrPUTsDCYxXWjP7clS81mZ6/803D8K4iM9Ma27VKas=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234015-3fc162c6f38a/go.mod h1:xURIpW9ES5+/GZhnV6beoEtxQrnkRGIfP5VQG2tCBLc=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19/go.mod h1:66JfowdXAEgad5O9NnYcsNPLCPZJD++2L9X0PCMODrA=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20240528184218-531527333157 h1:Zy9XzmMEflZ/MAaA7vNcoebnRAld7FsPW1EeBB7V0m8=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20240528184218-531527333157/go.mod h1:EfXuqaE1J41VCDicxHzUDm+8rk+7ZdXzHV0IhO/I6s0=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240521202816-d264139d666e h1:Elxv5MwEkCI9f5SkoL6afed6NTdxaGoAo39eANBwHL8=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240521202816-d264139d666e/go.mod h1:EfXuqaE1J41VCDicxHzUDm+8rk+7ZdXzHV0IhO/I6s0=
 google.golang.org/grpc v0.0.0-20160317175043-d3ddb4469d5a/go.mod h1:yo6s7OP7yaDglbqo1J04qKzAhqBH6lvTonzMVmEdcZw=
 google.golang.org/grpc v1.13.0/go.mod h1:yo6s7OP7yaDglbqo1J04qKzAhqBH6lvTonzMVmEdcZw=
 google.golang.org/grpc v1.14.0/go.mod h1:yo6s7OP7yaDglbqo1J04qKzAhqBH6lvTonzMVmEdcZw=
diff --git a/pkg/common/parameters.go b/pkg/common/parameters.go
index 3b07f077..2a1fa4d1 100644
--- a/pkg/common/parameters.go
+++ b/pkg/common/parameters.go
@@ -33,6 +33,7 @@ const (
 	ParameterKeyEnableConfidentialCompute     = "enable-confidential-storage"
 	ParameterKeyStoragePools                  = "storage-pools"
 	ParameterKeyResourceTags                  = "resource-tags"
+	ParameterKeyEnableMultiZoneProvisioning   = "enable-multi-zone-provisioning"
 
 	// Parameters for VolumeSnapshotClass
 	ParameterKeyStorageLocations = "storage-locations"
@@ -102,6 +103,9 @@ type DiskParameters struct {
 	// Values: {map[string]string}
 	// Default: ""
 	ResourceTags map[string]string
+	// Values: {bool}
+	// Default: false
+	MultiZoneProvisioning bool
 }
 
 // SnapshotParameters contains normalized and defaulted parameters for snapshots
@@ -121,11 +125,17 @@ type StoragePool struct {
 	ResourceName string
 }
 
+type ParameterProcessor struct {
+	DriverName         string
+	EnableStoragePools bool
+	EnableMultiZone    bool
+}
+
 // ExtractAndDefaultParameters will take the relevant parameters from a map and
 // put them into a well defined struct making sure to default unspecified fields.
 // extraVolumeLabels are added as labels; if there are also labels specified in
 // parameters, any matching extraVolumeLabels will be overridden.
-func ExtractAndDefaultParameters(parameters map[string]string, driverName string, extraVolumeLabels map[string]string, enableStoragePools bool, extraTags map[string]string) (DiskParameters, error) {
+func (pp *ParameterProcessor) ExtractAndDefaultParameters(parameters map[string]string, extraVolumeLabels map[string]string, extraTags map[string]string) (DiskParameters, error) {
 	p := DiskParameters{
 		DiskType:             "pd-standard",           // Default
 		ReplicationType:      replicationTypeNone,     // Default
@@ -210,24 +220,37 @@ func ExtractAndDefaultParameters(parameters map[string]string, driverName string
 
 			p.EnableConfidentialCompute = paramEnableConfidentialCompute
 		case ParameterKeyStoragePools:
-			if !enableStoragePools {
+			if !pp.EnableStoragePools {
 				return p, fmt.Errorf("parameters contains invalid option %q", ParameterKeyStoragePools)
 			}
 			storagePools, err := ParseStoragePools(v)
 			if err != nil {
-				return p, fmt.Errorf("parameters contain invalid value for %s parameter: %w", ParameterKeyStoragePools, err)
+				return p, fmt.Errorf("parameters contains invalid value for %s parameter %q: %w", ParameterKeyStoragePools, v, err)
 			}
 			p.StoragePools = storagePools
 		case ParameterKeyResourceTags:
 			if err := extractResourceTagsParameter(v, p.ResourceTags); err != nil {
 				return p, err
 			}
+		case ParameterKeyEnableMultiZoneProvisioning:
+			if !pp.EnableMultiZone {
+				return p, fmt.Errorf("parameters contains invalid option %q", ParameterKeyEnableMultiZoneProvisioning)
+			}
+			paramEnableMultiZoneProvisioning, err := ConvertStringToBool(v)
+			if err != nil {
+				return p, fmt.Errorf("parameters contain invalid value for %s parameter: %w", ParameterKeyEnableMultiZoneProvisioning, err)
+			}
+
+			p.MultiZoneProvisioning = paramEnableMultiZoneProvisioning
+			if paramEnableMultiZoneProvisioning {
+				p.Labels[MultiZoneLabel] = "true"
+			}
 		default:
 			return p, fmt.Errorf("parameters contains invalid option %q", k)
 		}
 	}
 	if len(p.Tags) > 0 {
-		p.Tags[tagKeyCreatedBy] = driverName
+		p.Tags[tagKeyCreatedBy] = pp.DriverName
 	}
 	return p, nil
 }
diff --git a/pkg/common/utils.go b/pkg/common/utils.go
index 25f8a655..c3b71c1c 100644
--- a/pkg/common/utils.go
+++ b/pkg/common/utils.go
@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"net/http"
 	"regexp"
+	"slices"
 	"strings"
 	"time"
 
@@ -73,6 +74,10 @@ const (
 	// Full or partial URL of the machine type resource, in the format:
 	//   zones/zone/machineTypes/machine-type
 	machineTypePattern = "zones/[^/]+/machineTypes/([^/]+)$"
+
+	// Full or partial URL of the zone resource, in the format:
+	//   projects/{project}/zones/{zone}
+	zoneURIPattern = "projects/[^/]+/zones/([^/]+)$"
 )
 
 var (
@@ -85,6 +90,8 @@ var (
 
 	storagePoolFieldsRegex = regexp.MustCompile(`^projects/([^/]+)/zones/([^/]+)/storagePools/([^/]+)$`)
 
+	zoneURIRegex = regexp.MustCompile(zoneURIPattern)
+
 	// userErrorCodeMap tells how API error types are translated to error codes.
 	userErrorCodeMap = map[int]codes.Code{
 		http.StatusForbidden:       codes.PermissionDenied,
@@ -97,6 +104,8 @@ var (
 	regexParent = regexp.MustCompile(`(^[1-9][0-9]{0,31}$)|(^[a-z][a-z0-9-]{4,28}[a-z0-9]$)`)
 	regexKey    = regexp.MustCompile(`^[a-zA-Z0-9]([0-9A-Za-z_.-]{0,61}[a-zA-Z0-9])?$`)
 	regexValue  = regexp.MustCompile(`^[a-zA-Z0-9]([0-9A-Za-z_.@%=+:,*#&()\[\]{}\-\s]{0,61}[a-zA-Z0-9])?$`)
+
+	csiRetryableErrorCodes = []codes.Code{codes.Canceled, codes.DeadlineExceeded, codes.Unavailable, codes.Aborted, codes.ResourceExhausted}
 )
 
 func BytesToGbRoundDown(bytes int64) int64 {
@@ -545,9 +554,37 @@ func isGoogleAPIError(err error) (codes.Code, error) {
 	return codes.Unknown, fmt.Errorf("googleapi.Error %w does not map to any known errors", err)
 }
 
-func LoggedError(msg string, err error) error {
+func loggedErrorForCode(msg string, code codes.Code, err error) error {
 	klog.Errorf(msg+"%v", err.Error())
-	return status.Errorf(CodeForError(err), msg+"%v", err.Error())
+	return status.Errorf(code, msg+"%v", err.Error())
+}
+
+func LoggedError(msg string, err error) error {
+	return loggedErrorForCode(msg, CodeForError(err), err)
+}
+
+// NewCombinedError tries to return an appropriate wrapped error that captures
+// useful information as an error code
+// If there are multiple errors, it extracts the first "retryable" error
+// as interpreted by the CSI sidecar.
+func NewCombinedError(msg string, errs []error) error {
+	// If there is only one error, return it as the single error code
+	if len(errs) == 1 {
+		LoggedError(msg, errs[0])
+	}
+
+	for _, err := range errs {
+		code := CodeForError(err)
+		if slices.Contains(csiRetryableErrorCodes, code) {
+			// Return this as a TemporaryError to lock-in the retryable code
+			// This will invoke the "existing" error code check in CodeForError
+			return NewTemporaryError(code, fmt.Errorf("%s: %w", msg, err))
+		}
+	}
+
+	// None of these error codes were retryable. Just return a combined error
+	// The first matching error (based on our CodeForError) logic will be returned.
+	return LoggedError(msg, errors.Join(errs...))
 }
 
 func isValidDiskEncryptionKmsKey(DiskEncryptionKmsKey string) bool {
@@ -556,6 +593,14 @@ func isValidDiskEncryptionKmsKey(DiskEncryptionKmsKey string) bool {
 	return kmsKeyPattern.MatchString(DiskEncryptionKmsKey)
 }
 
+func ParseZoneFromURI(zoneURI string) (string, error) {
+	zoneMatch := zoneURIRegex.FindStringSubmatch(zoneURI)
+	if zoneMatch == nil {
+		return "", fmt.Errorf("failed to parse zone URI. Expected projects/{project}/zones/{zone}. Got: %s", zoneURI)
+	}
+	return zoneMatch[1], nil
+}
+
 // ParseStoragePools returns an error if none of the given storagePools
 // (delimited by a comma) are in the format
 // projects/project/zones/zone/storagePools/storagePool.
diff --git a/pkg/gce-cloud-provider/compute/cloud-disk.go b/pkg/gce-cloud-provider/compute/cloud-disk.go
index bbd3afc1..02b70d5d 100644
--- a/pkg/gce-cloud-provider/compute/cloud-disk.go
+++ b/pkg/gce-cloud-provider/compute/cloud-disk.go
@@ -256,3 +256,14 @@ func (d *CloudDisk) GetLabels() map[string]string {
 		return nil
 	}
 }
+
+func (d *CloudDisk) GetAccessMode() string {
+	switch {
+	case d.disk != nil:
+		return d.disk.AccessMode
+	case d.betaDisk != nil:
+		return d.betaDisk.AccessMode
+	default:
+		return ""
+	}
+}
diff --git a/pkg/gce-cloud-provider/compute/fake-gce.go b/pkg/gce-cloud-provider/compute/fake-gce.go
index 3b7416b1..77dbf075 100644
--- a/pkg/gce-cloud-provider/compute/fake-gce.go
+++ b/pkg/gce-cloud-provider/compute/fake-gce.go
@@ -79,7 +79,11 @@ func CreateFakeCloudProvider(project, zone string, cloudDisks []*CloudDisk) (*Fa
 		mockDiskStatus: "READY",
 	}
 	for _, d := range cloudDisks {
-		fcp.disks[d.GetName()] = d
+		diskZone := d.GetZone()
+		if diskZone == "" {
+			diskZone = zone
+		}
+		fcp.disks[meta.ZonalKey(d.GetName(), diskZone).String()] = d
 	}
 	return fcp, nil
 }
@@ -101,8 +105,8 @@ func (cloud *FakeCloudProvider) RepairUnderspecifiedVolumeKey(ctx context.Contex
 		if volumeKey.Zone != common.UnspecifiedValue {
 			return project, volumeKey, nil
 		}
-		for name, d := range cloud.disks {
-			if name == volumeKey.Name {
+		for diskVolKey, d := range cloud.disks {
+			if diskVolKey == volumeKey.String() {
 				volumeKey.Zone = d.GetZone()
 				return project, volumeKey, nil
 			}
@@ -127,6 +131,15 @@ func (cloud *FakeCloudProvider) ListZones(ctx context.Context, region string) ([
 	return []string{cloud.zone, "country-region-fakesecondzone"}, nil
 }
 
+func (cloud *FakeCloudProvider) ListCompatibleDiskTypeZones(ctx context.Context, project string, zones []string, diskType string) ([]string, error) {
+	// Assume all zones are compatible
+	return zones, nil
+}
+
+func (cloud *FakeCloudProvider) ListDisksWithFilter(ctx context.Context, fields []googleapi.Field, filter string) ([]*computev1.Disk, string, error) {
+	return cloud.ListDisks(ctx, fields)
+}
+
 func (cloud *FakeCloudProvider) ListDisks(ctx context.Context, fields []googleapi.Field) ([]*computev1.Disk, string, error) {
 	d := []*computev1.Disk{}
 	for _, cd := range cloud.disks {
@@ -167,7 +180,7 @@ func (cloud *FakeCloudProvider) ListSnapshots(ctx context.Context, filter string
 
 // Disk Methods
 func (cloud *FakeCloudProvider) GetDisk(ctx context.Context, project string, volKey *meta.Key, api GCEAPIVersion) (*CloudDisk, error) {
-	disk, ok := cloud.disks[volKey.Name]
+	disk, ok := cloud.disks[volKey.String()]
 	if !ok {
 		return nil, notFoundError()
 	}
@@ -203,8 +216,8 @@ func (cloud *FakeCloudProvider) ValidateExistingDisk(ctx context.Context, resp *
 	return ValidateDiskParameters(resp, params)
 }
 
-func (cloud *FakeCloudProvider) InsertDisk(ctx context.Context, project string, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool) error {
-	if disk, ok := cloud.disks[volKey.Name]; ok {
+func (cloud *FakeCloudProvider) InsertDisk(ctx context.Context, project string, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool, accessMode string) error {
+	if disk, ok := cloud.disks[volKey.String()]; ok {
 		err := cloud.ValidateExistingDisk(ctx, disk, params,
 			int64(capacityRange.GetRequiredBytes()),
 			int64(capacityRange.GetLimitBytes()),
@@ -259,15 +272,15 @@ func (cloud *FakeCloudProvider) InsertDisk(ctx context.Context, project string,
 	if containsBetaDiskType(hyperdiskTypes, params.DiskType) {
 		betaDisk := convertV1DiskToBetaDisk(computeDisk)
 		betaDisk.EnableConfidentialCompute = params.EnableConfidentialCompute
-		cloud.disks[volKey.Name] = CloudDiskFromBeta(betaDisk)
+		cloud.disks[volKey.String()] = CloudDiskFromBeta(betaDisk)
 	} else {
-		cloud.disks[volKey.Name] = CloudDiskFromV1(computeDisk)
+		cloud.disks[volKey.String()] = CloudDiskFromV1(computeDisk)
 	}
 	return nil
 }
 
 func (cloud *FakeCloudProvider) DeleteDisk(ctx context.Context, project string, volKey *meta.Key) error {
-	delete(cloud.disks, volKey.Name)
+	delete(cloud.disks, volKey.String())
 	return nil
 }
 
@@ -307,6 +320,22 @@ func (cloud *FakeCloudProvider) DetachDisk(ctx context.Context, project, deviceN
 	return nil
 }
 
+func (cloud *FakeCloudProvider) SetDiskAccessMode(ctx context.Context, project string, volKey *meta.Key, accessMode string) error {
+	disk, ok := cloud.disks[volKey.String()]
+	if !ok {
+		return fmt.Errorf("disk %v not found", volKey)
+	}
+
+	if disk.disk != nil {
+		disk.disk.AccessMode = accessMode
+	}
+	if disk.betaDisk != nil {
+		disk.betaDisk.AccessMode = accessMode
+	}
+
+	return nil
+}
+
 func (cloud *FakeCloudProvider) GetDiskTypeURI(project string, volKey *meta.Key, diskType string) string {
 	switch volKey.Type() {
 	case meta.Zonal:
@@ -390,7 +419,7 @@ func (cloud *FakeCloudProvider) CreateSnapshot(ctx context.Context, project stri
 }
 
 func (cloud *FakeCloudProvider) ResizeDisk(ctx context.Context, project string, volKey *meta.Key, requestBytes int64) (int64, error) {
-	disk, ok := cloud.disks[volKey.Name]
+	disk, ok := cloud.disks[volKey.String()]
 	if !ok {
 		return -1, notFoundError()
 	}
diff --git a/pkg/gce-cloud-provider/compute/gce-compute.go b/pkg/gce-cloud-provider/compute/gce-compute.go
index eaa0c995..13f00fcc 100644
--- a/pkg/gce-cloud-provider/compute/gce-compute.go
+++ b/pkg/gce-cloud-provider/compute/gce-compute.go
@@ -39,6 +39,7 @@ import (
 	"google.golang.org/grpc/status"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/klog/v2"
+	"k8s.io/utils/strings/slices"
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
 )
 
@@ -64,6 +65,8 @@ const (
 	GCEAPIVersionBeta GCEAPIVersion = "beta"
 )
 
+var GCEAPIVersions = []GCEAPIVersion{GCEAPIVersionBeta, GCEAPIVersionV1}
+
 // AttachDiskBackoff is backoff used to wait for AttachDisk to complete.
 // Default values are similar to Poll every 5 seconds with 2 minute timeout.
 var AttachDiskBackoff = wait.Backoff{
@@ -99,15 +102,18 @@ type GCECompute interface {
 	GetDisk(ctx context.Context, project string, volumeKey *meta.Key, gceAPIVersion GCEAPIVersion) (*CloudDisk, error)
 	RepairUnderspecifiedVolumeKey(ctx context.Context, project string, volumeKey *meta.Key) (string, *meta.Key, error)
 	ValidateExistingDisk(ctx context.Context, disk *CloudDisk, params common.DiskParameters, reqBytes, limBytes int64, multiWriter bool) error
-	InsertDisk(ctx context.Context, project string, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool) error
+	InsertDisk(ctx context.Context, project string, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool, accessMode string) error
 	DeleteDisk(ctx context.Context, project string, volumeKey *meta.Key) error
 	AttachDisk(ctx context.Context, project string, volKey *meta.Key, readWrite, diskType, instanceZone, instanceName string, forceAttach bool) error
 	DetachDisk(ctx context.Context, project, deviceName, instanceZone, instanceName string) error
+	SetDiskAccessMode(ctx context.Context, project string, volKey *meta.Key, accessMode string) error
+	ListCompatibleDiskTypeZones(ctx context.Context, project string, zones []string, diskType string) ([]string, error)
 	GetDiskSourceURI(project string, volKey *meta.Key) string
 	GetDiskTypeURI(project string, volKey *meta.Key, diskType string) string
 	WaitForAttach(ctx context.Context, project string, volKey *meta.Key, diskType, instanceZone, instanceName string) error
 	ResizeDisk(ctx context.Context, project string, volKey *meta.Key, requestBytes int64) (int64, error)
 	ListDisks(ctx context.Context, fields []googleapi.Field) ([]*computev1.Disk, string, error)
+	ListDisksWithFilter(ctx context.Context, fields []googleapi.Field, filter string) ([]*computev1.Disk, string, error)
 	ListInstances(ctx context.Context, fields []googleapi.Field) ([]*computev1.Instance, string, error)
 	// Regional Disk Methods
 	GetReplicaZoneURI(project string, zone string) string
@@ -138,6 +144,15 @@ func (cloud *CloudProvider) GetDefaultZone() string {
 // ListDisks lists disks based on maxEntries and pageToken only in the project
 // and region that the driver is running in.
 func (cloud *CloudProvider) ListDisks(ctx context.Context, fields []googleapi.Field) ([]*computev1.Disk, string, error) {
+	filter := ""
+	return cloud.listDisksInternal(ctx, fields, filter)
+}
+
+func (cloud *CloudProvider) ListDisksWithFilter(ctx context.Context, fields []googleapi.Field, filter string) ([]*computev1.Disk, string, error) {
+	return cloud.listDisksInternal(ctx, fields, filter)
+}
+
+func (cloud *CloudProvider) listDisksInternal(ctx context.Context, fields []googleapi.Field, filter string) ([]*computev1.Disk, string, error) {
 	region, err := common.GetRegionFromZones([]string{cloud.zone})
 	if err != nil {
 		return nil, "", fmt.Errorf("failed to get region from zones: %w", err)
@@ -151,6 +166,7 @@ func (cloud *CloudProvider) ListDisks(ctx context.Context, fields []googleapi.Fi
 	// listing out regional disks in the region
 	rlCall := cloud.service.RegionDisks.List(cloud.project, region)
 	rlCall.Fields(fields...)
+	rlCall.Filter(filter)
 	nextPageToken := "pageToken"
 	for nextPageToken != "" {
 		rDiskList, err := rlCall.Do()
@@ -166,6 +182,7 @@ func (cloud *CloudProvider) ListDisks(ctx context.Context, fields []googleapi.Fi
 	for _, zone := range zones {
 		lCall := cloud.service.Disks.List(cloud.project, zone)
 		lCall.Fields(fields...)
+		lCall.Filter(filter)
 		nextPageToken := "pageToken"
 		for nextPageToken != "" {
 			diskList, err := lCall.Do()
@@ -418,7 +435,7 @@ func ValidateDiskParameters(disk *CloudDisk, params common.DiskParameters) error
 	return nil
 }
 
-func (cloud *CloudProvider) InsertDisk(ctx context.Context, project string, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool) error {
+func (cloud *CloudProvider) InsertDisk(ctx context.Context, project string, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool, accessMode string) error {
 	klog.V(5).Infof("Inserting disk %v", volKey)
 
 	description, err := encodeTags(params.Tags)
@@ -431,7 +448,7 @@ func (cloud *CloudProvider) InsertDisk(ctx context.Context, project string, volK
 		if description == "" {
 			description = "Disk created by GCE-PD CSI Driver"
 		}
-		return cloud.insertZonalDisk(ctx, project, volKey, params, capBytes, capacityRange, snapshotID, volumeContentSourceVolumeID, description, multiWriter)
+		return cloud.insertZonalDisk(ctx, project, volKey, params, capBytes, capacityRange, snapshotID, volumeContentSourceVolumeID, description, multiWriter, accessMode)
 	case meta.Regional:
 		if description == "" {
 			description = "Regional disk created by GCE-PD CSI Driver"
@@ -492,6 +509,7 @@ func convertV1DiskToBetaDisk(v1Disk *computev1.Disk) *computebeta.Disk {
 		Status:            v1Disk.Status,
 		SelfLink:          v1Disk.SelfLink,
 		Params:            params,
+		AccessMode:        v1Disk.AccessMode,
 	}
 
 	// Hyperdisk doesn't currently support multiWriter (https://cloud.google.com/compute/docs/disks/hyperdisks#limitations),
@@ -646,7 +664,8 @@ func (cloud *CloudProvider) insertZonalDisk(
 	snapshotID string,
 	volumeContentSourceVolumeID string,
 	description string,
-	multiWriter bool) error {
+	multiWriter bool,
+	accessMode string) error {
 	var (
 		err           error
 		opName        string
@@ -714,6 +733,7 @@ func (cloud *CloudProvider) insertZonalDisk(
 			ResourceManagerTags: resourceTags,
 		}
 	}
+	diskToCreate.AccessMode = accessMode
 
 	if gceAPIVersion == GCEAPIVersionBeta {
 		var insertOp *computebeta.Operation
@@ -875,6 +895,73 @@ func (cloud *CloudProvider) DetachDisk(ctx context.Context, project, deviceName,
 	return nil
 }
 
+func (cloud *CloudProvider) SetDiskAccessMode(ctx context.Context, project string, volKey *meta.Key, accessMode string) error {
+	diskMask := &computev1.Disk{
+		AccessMode: accessMode,
+		Name:       volKey.Name,
+	}
+	switch volKey.Type() {
+	case meta.Zonal:
+		op, err := cloud.service.Disks.Update(project, volKey.Zone, volKey.Name, diskMask).Context(ctx).Paths("accessMode").Do()
+		if err != nil {
+			return fmt.Errorf("failed to set access mode for zonal volume %v: %w", volKey, err)
+		}
+		klog.V(5).Infof("SetDiskAccessMode operation %s for disk %s", op.Name, volKey.Name)
+
+		err = cloud.waitForZonalOp(ctx, project, op.Name, volKey.Zone)
+		if err != nil {
+			return fmt.Errorf("failed waiting for op for zonal disk update for %v: %w", volKey, err)
+		}
+	case meta.Regional:
+		op, err := cloud.service.RegionDisks.Update(project, volKey.Region, volKey.Name, diskMask).Context(ctx).Paths("accessMode").Do()
+		if err != nil {
+			return fmt.Errorf("failed to set access mode for regional volume %v: %w", volKey, err)
+		}
+		klog.V(5).Infof("SetDiskAccessMode operation %s for disk %s", op.Name, volKey.Name)
+
+		err = cloud.waitForRegionalOp(ctx, project, op.Name, volKey.Region)
+		if err != nil {
+			return fmt.Errorf("failed waiting for op for regional disk update for %v: %w", volKey, err)
+		}
+	default:
+		return fmt.Errorf("volume key %v not zonal nor regional", volKey.Name)
+	}
+
+	return nil
+}
+
+func (cloud *CloudProvider) ListCompatibleDiskTypeZones(ctx context.Context, project string, zones []string, diskType string) ([]string, error) {
+	diskTypeFilter := fmt.Sprintf("name=%s", diskType)
+	filters := []string{diskTypeFilter}
+	diskTypeListCall := cloud.service.DiskTypes.AggregatedList(project).Context(ctx).Filter(strings.Join(filters, " "))
+
+	supportedZones := []string{}
+	nextPageToken := "pageToken"
+	for nextPageToken != "" {
+		diskTypeList, err := diskTypeListCall.Do()
+		if err != nil {
+			return nil, err
+		}
+		for _, item := range diskTypeList.Items {
+			for _, diskType := range item.DiskTypes {
+				zone, err := common.ParseZoneFromURI(diskType.Zone)
+				if err != nil {
+					klog.Warningf("Failed to parse zone %q from diskTypes API: %v", diskType.Zone, err)
+					continue
+				}
+				if slices.Contains(zones, zone) {
+					supportedZones = append(supportedZones, zone)
+				}
+			}
+		}
+
+		nextPageToken = diskTypeList.NextPageToken
+		diskTypeListCall.PageToken(nextPageToken)
+	}
+
+	return supportedZones, nil
+}
+
 func (cloud *CloudProvider) GetDiskSourceURI(project string, volKey *meta.Key) string {
 	switch volKey.Type() {
 	case meta.Zonal:
diff --git a/pkg/gce-cloud-provider/compute/gce.go b/pkg/gce-cloud-provider/compute/gce.go
index c2d2df8c..f25519cf 100644
--- a/pkg/gce-cloud-provider/compute/gce.go
+++ b/pkg/gce-cloud-provider/compute/gce.go
@@ -42,7 +42,6 @@ import (
 )
 
 type Environment string
-type Version string
 
 const (
 	TokenURL                        = "https://accounts.google.com/o/oauth2/token"
@@ -54,8 +53,6 @@ const (
 	regionURITemplate = "projects/%s/regions/%s"
 
 	replicaZoneURITemplateSingleZone             = "projects/%s/zones/%s" // {gce.projectID}/zones/{disk.Zone}
-	versionV1                        Version     = "v1"
-	versionBeta                      Version     = "beta"
 	EnvironmentStaging               Environment = "staging"
 	EnvironmentProduction            Environment = "production"
 
@@ -229,7 +226,7 @@ func readConfig(configPath string) (*ConfigFile, error) {
 }
 
 func createBetaCloudService(ctx context.Context, vendorVersion string, tokenSource oauth2.TokenSource, computeEndpoint *url.URL, computeEnvironment Environment) (*computebeta.Service, error) {
-	computeOpts, err := getComputeVersion(ctx, tokenSource, computeEndpoint, computeEnvironment, versionBeta)
+	computeOpts, err := getComputeVersion(ctx, tokenSource, computeEndpoint, computeEnvironment, GCEAPIVersionBeta)
 	if err != nil {
 		klog.Errorf("Failed to get compute endpoint: %s", err)
 	}
@@ -242,7 +239,7 @@ func createBetaCloudService(ctx context.Context, vendorVersion string, tokenSour
 }
 
 func createCloudService(ctx context.Context, vendorVersion string, tokenSource oauth2.TokenSource, computeEndpoint *url.URL, computeEnvironment Environment) (*compute.Service, error) {
-	computeOpts, err := getComputeVersion(ctx, tokenSource, computeEndpoint, computeEnvironment, versionV1)
+	computeOpts, err := getComputeVersion(ctx, tokenSource, computeEndpoint, computeEnvironment, GCEAPIVersionV1)
 	if err != nil {
 		klog.Errorf("Failed to get compute endpoint: %s", err)
 	}
@@ -254,7 +251,7 @@ func createCloudService(ctx context.Context, vendorVersion string, tokenSource o
 	return service, nil
 }
 
-func getComputeVersion(ctx context.Context, tokenSource oauth2.TokenSource, computeEndpoint *url.URL, computeEnvironment Environment, computeVersion Version) ([]option.ClientOption, error) {
+func getComputeVersion(ctx context.Context, tokenSource oauth2.TokenSource, computeEndpoint *url.URL, computeEnvironment Environment, computeVersion GCEAPIVersion) ([]option.ClientOption, error) {
 	client, err := newOauthClient(ctx, tokenSource)
 	if err != nil {
 		return nil, err
@@ -270,7 +267,7 @@ func getComputeVersion(ctx context.Context, tokenSource oauth2.TokenSource, comp
 	return computeOpts, nil
 }
 
-func constructComputeEndpointPath(env Environment, version Version) string {
+func constructComputeEndpointPath(env Environment, version GCEAPIVersion) string {
 	prefix := ""
 	if env == EnvironmentStaging {
 		prefix = fmt.Sprintf("%s_", env)
diff --git a/pkg/gce-pd-csi-driver/controller.go b/pkg/gce-pd-csi-driver/controller.go
index 81c0426a..a2d57000 100644
--- a/pkg/gce-pd-csi-driver/controller.go
+++ b/pkg/gce-pd-csi-driver/controller.go
@@ -195,6 +195,8 @@ const (
 	resourceProject    = "projects"
 
 	listDisksUsersField = googleapi.Field("items/users")
+
+	readOnlyManyAccessMode = "READ_ONLY_MANY"
 )
 
 var (
@@ -214,7 +216,8 @@ var (
 		"items/selfLink",
 		"nextPageToken",
 	}
-	listDisksFieldsWithUsers = append(listDisksFieldsWithoutUsers, "items/users")
+	listDisksFieldsWithUsers      = append(listDisksFieldsWithoutUsers, "items/users")
+	disksWithModifiableAccessMode = []string{"hyperdisk-ml"}
 )
 
 func isDiskReady(disk *gce.CloudDisk) (bool, error) {
@@ -279,6 +282,15 @@ func useVolumeCloning(req *csi.CreateVolumeRequest) bool {
 }
 
 func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
+	response, err := gceCS.createVolumeInternal(ctx, req)
+	if err != nil && req != nil {
+		klog.V(4).Infof("CreateVolume succeeded for volume %v", req.Name)
+	}
+
+	return response, err
+}
+
+func (gceCS *GCEControllerServer) createVolumeInternal(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
 	var err error
 	diskTypeForMetric := metrics.DefaultDiskTypeForMetric
 	enableConfidentialCompute := metrics.DefaultEnableConfidentialCompute
@@ -288,17 +300,16 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 	}()
 	// Validate arguments
 	volumeCapabilities := req.GetVolumeCapabilities()
-	name := req.GetName()
 	capacityRange := req.GetCapacityRange()
-	if len(name) == 0 {
+	if len(req.GetName()) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "CreateVolume Name must be provided")
 	}
 	if volumeCapabilities == nil || len(volumeCapabilities) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "CreateVolume Volume capabilities must be provided")
 	}
 
-	capBytes, err := getRequestCapacity(capacityRange)
-	if err != nil {
+	// Validate request capacity early
+	if _, err := getRequestCapacity(capacityRange); err != nil {
 		return nil, status.Errorf(codes.InvalidArgument, "CreateVolume Request Capacity is invalid: %v", err.Error())
 	}
 
@@ -309,7 +320,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 
 	// Apply Parameters (case-insensitive). We leave validation of
 	// the values to the cloud provider.
-	params, err := common.ExtractAndDefaultParameters(req.GetParameters(), gceCS.Driver.name, gceCS.Driver.extraVolumeLabels, gceCS.enableStoragePools, gceCS.Driver.extraTags)
+	params, err := gceCS.parameterProcessor().ExtractAndDefaultParameters(req.GetParameters(), gceCS.Driver.extraVolumeLabels, gceCS.Driver.extraTags)
 	diskTypeForMetric = params.DiskType
 	enableConfidentialCompute = strconv.FormatBool(params.EnableConfidentialCompute)
 	hasStoragePools := len(params.StoragePools) > 0
@@ -317,11 +328,9 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 	if err != nil {
 		return nil, status.Errorf(codes.InvalidArgument, "failed to extract parameters: %v", err.Error())
 	}
-	// Determine multiWriter
-	gceAPIVersion := gce.GCEAPIVersionV1
-	multiWriter, _ := getMultiWriterFromCapabilities(volumeCapabilities)
-	if multiWriter {
-		gceAPIVersion = gce.GCEAPIVersionBeta
+	// Validate multiwriter
+	if _, err := getMultiWriterFromCapabilities(volumeCapabilities); err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "VolumeCapabilities is invalid: %v", err.Error())
 	}
 
 	err = validateStoragePools(req, params, gceCS.CloudProvider.GetDefaultProject())
@@ -331,11 +340,176 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 		return nil, err
 	}
 
+	// Validate VolumeContentSource is set when access mode is read only
+	readonly, _ := getReadOnlyFromCapabilities(volumeCapabilities)
+	if readonly && req.GetVolumeContentSource() == nil {
+		return nil, status.Error(codes.InvalidArgument, "VolumeContentSource must be provided when AccessMode is set to read only")
+	}
+
+	// Validate multi-zone provisioning configuration
+	err = gceCS.validateMultiZoneProvisioning(req, params)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "CreateVolume failed to validate multi-zone provisioning request: %v", err)
+	}
+
 	// Verify that the regional availability class is only used on regional disks.
 	if params.ForceAttach && params.ReplicationType != replicationTypeRegionalPD {
 		return nil, status.Errorf(codes.InvalidArgument, "invalid availabilty class for zonal disk")
 	}
 
+	if gceCS.multiZoneVolumeHandleConfig.Enable && params.MultiZoneProvisioning {
+		// Create multi-zone disk, that may have up to N disks.
+		return gceCS.createMultiZoneDisk(ctx, req, params)
+	}
+
+	// Create single device zonal or regional disk
+	return gceCS.createSingleDeviceDisk(ctx, req, params)
+}
+
+func (gceCS *GCEControllerServer) getSupportedZonesForPDType(ctx context.Context, zones []string, diskType string) ([]string, error) {
+	project := gceCS.CloudProvider.GetDefaultProject()
+	zones, err := gceCS.CloudProvider.ListCompatibleDiskTypeZones(ctx, project, zones, diskType)
+	if err != nil {
+		return nil, err
+	}
+	return zones, nil
+}
+
+func (gceCS *GCEControllerServer) getMultiZoneProvisioningZones(ctx context.Context, req *csi.CreateVolumeRequest, params common.DiskParameters) ([]string, error) {
+	top := req.GetAccessibilityRequirements()
+	if top == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "no topology specified")
+	}
+	prefZones, err := getZonesFromTopology(top.GetPreferred())
+	if err != nil {
+		return nil, fmt.Errorf("could not get zones from preferred topology: %w", err)
+	}
+	reqZones, err := getZonesFromTopology(top.GetRequisite())
+	if err != nil {
+		return nil, fmt.Errorf("could not get zones from requisite topology: %w", err)
+	}
+	prefSet := sets.NewString(prefZones...)
+	reqSet := sets.NewString(reqZones...)
+	prefAndReqSet := prefSet.Union(reqSet)
+	availableZones := prefAndReqSet.List()
+	if prefAndReqSet.Len() == 0 {
+		// If there are no specified zones, this means that there were no aggregate
+		// zones (eg: no nodes running) in the cluster
+		availableZones = gceCS.fallbackRequisiteZones
+	}
+
+	supportedZones, err := gceCS.getSupportedZonesForPDType(ctx, availableZones, params.DiskType)
+	if err != nil {
+		return nil, fmt.Errorf("could not get supported zones for disk type %v from zone list %v: %w", params.DiskType, prefAndReqSet.List(), err)
+	}
+
+	// It's possible that the provided requisite zones shifted since the last time that
+	// CreateVolume was called (eg: due to a node being removed in a zone)
+	// Ensure that we combine the supportedZones with any existing zones to get the full set.
+	existingZones, err := gceCS.getZonesWithDiskNameAndType(ctx, req.Name, params.DiskType)
+	if err != nil {
+		return nil, common.LoggedError(fmt.Sprintf("failed to check existing list of zones for request: %v", req.Name), err)
+	}
+
+	supportedSet := sets.NewString(supportedZones...)
+	existingSet := sets.NewString(existingZones...)
+	combinedZones := existingSet.Union(supportedSet)
+
+	return combinedZones.List(), nil
+}
+
+func (gceCS *GCEControllerServer) createMultiZoneDisk(ctx context.Context, req *csi.CreateVolumeRequest, params common.DiskParameters) (*csi.CreateVolumeResponse, error) {
+	// Determine the zones that are needed.
+	var err error
+
+	// For multi-zone, we either select:
+	// 1) The zones specified in requisite topology requirements
+	// 2) All zones in the region that are compatible with the selected disk type
+	zones, err := gceCS.getMultiZoneProvisioningZones(ctx, req, params)
+	if err != nil {
+		return nil, err
+	}
+
+	multiZoneVolKey := meta.ZonalKey(req.GetName(), common.MultiZoneValue)
+	volumeID, err := common.KeyToVolumeID(multiZoneVolKey, gceCS.CloudProvider.GetDefaultProject())
+	if err != nil {
+		return nil, err
+	}
+	if acquired := gceCS.volumeLocks.TryAcquire(volumeID); !acquired {
+		return nil, status.Errorf(codes.Aborted, common.VolumeOperationAlreadyExistsFmt, volumeID)
+	}
+	defer gceCS.volumeLocks.Release(volumeID)
+
+	createDiskErrs := []error{}
+	createdDisks := make([]*gce.CloudDisk, 0, len(zones))
+	for _, zone := range zones {
+		volKey := meta.ZonalKey(req.GetName(), zone)
+		klog.V(4).Infof("Creating single zone disk for zone %q and volume: %v", zone, volKey)
+		disk, err := gceCS.createSingleDisk(ctx, req, params, volKey, []string{zone})
+		if err != nil {
+			createDiskErrs = append(createDiskErrs, err)
+			continue
+		}
+
+		createdDisks = append(createdDisks, disk)
+	}
+	if len(createDiskErrs) > 0 {
+		return nil, common.LoggedError("Failed to create multi-zone disk: ", errors.Join(createDiskErrs...))
+	}
+
+	if len(createdDisks) == 0 {
+		return nil, status.Errorf(codes.Internal, "could not create any disks for request: %v", req)
+	}
+
+	// Use the first response as a template
+	volumeId := fmt.Sprintf("projects/%s/zones/%s/disks/%s", gceCS.CloudProvider.GetDefaultProject(), common.MultiZoneValue, req.GetName())
+	klog.V(4).Infof("CreateVolume succeeded for multi-zone disks in zones %s: %v", zones, multiZoneVolKey)
+	return generateCreateVolumeResponseWithVolumeId(createdDisks[0], zones, params, volumeId), nil
+}
+
+func (gceCS *GCEControllerServer) getZonesWithDiskNameAndType(ctx context.Context, name string, diskType string) ([]string, error) {
+	zoneOnlyFields := []googleapi.Field{"items/zone", "items/type"}
+	nameAndRegionFilter := fmt.Sprintf("name=%s", name)
+	disksWithZone, _, err := gceCS.CloudProvider.ListDisksWithFilter(ctx, zoneOnlyFields, nameAndRegionFilter)
+	if err != nil {
+		return nil, fmt.Errorf("failed to check existing zones for disk name %v: %w", name, err)
+	}
+	zones := []string{}
+	for _, disk := range disksWithZone {
+		if !strings.Contains(disk.Type, diskType) || disk.Zone == "" {
+			continue
+		}
+		diskZone, err := common.ParseZoneFromURI(disk.Zone)
+		if err != nil {
+			klog.Warningf("Malformed zone URI %v for disk %v from ListDisks call. Skipping", disk.Zone, name)
+			continue
+		}
+		zones = append(zones, diskZone)
+	}
+	return zones, nil
+}
+
+func (gceCS *GCEControllerServer) updateAccessModeIfNecessary(ctx context.Context, volKey *meta.Key, disk *gce.CloudDisk, readonly bool) error {
+	if !slices.Contains(disksWithModifiableAccessMode, disk.GetPDType()) {
+		// If this isn't a disk that has access mode (eg: Hyperdisk ML), return
+		// So far, HyperdiskML is the only disk type that allows the disk type to be modified.
+		return nil
+	}
+	if !readonly {
+		// Only update the access mode if we're converting from ReadWrite to ReadOnly
+		return nil
+	}
+	project := gceCS.CloudProvider.GetDefaultProject()
+	if disk.GetAccessMode() == readOnlyManyAccessMode {
+		// If the access mode is already readonly, return
+		return nil
+	}
+
+	return gceCS.CloudProvider.SetDiskAccessMode(ctx, project, volKey, readOnlyManyAccessMode)
+}
+
+func (gceCS *GCEControllerServer) createSingleDeviceDisk(ctx context.Context, req *csi.CreateVolumeRequest, params common.DiskParameters) (*csi.CreateVolumeResponse, error) {
+	var err error
 	var locationTopReq *locationRequirements
 	if useVolumeCloning(req) {
 		locationTopReq, err = cloningLocationRequirements(req, params.ReplicationType)
@@ -356,7 +530,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 		if len(zones) != 1 {
 			return nil, status.Errorf(codes.Internal, "Failed to pick exactly 1 zone for zonal disk, got %v instead", len(zones))
 		}
-		volKey = meta.ZonalKey(name, zones[0])
+		volKey = meta.ZonalKey(req.GetName(), zones[0])
 
 	case replicationTypeRegionalPD:
 		zones, err = gceCS.pickZones(ctx, req.GetAccessibilityRequirements(), 2, locationTopReq)
@@ -367,7 +541,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 		if err != nil {
 			return nil, status.Errorf(codes.InvalidArgument, "CreateVolume failed to get region from zones: %v", err.Error())
 		}
-		volKey = meta.RegionalKey(name, region)
+		volKey = meta.RegionalKey(req.GetName(), region)
 	default:
 		return nil, status.Errorf(codes.InvalidArgument, "CreateVolume replication type '%s' is not supported", params.ReplicationType)
 	}
@@ -380,9 +554,27 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 		return nil, status.Errorf(codes.Aborted, common.VolumeOperationAlreadyExistsFmt, volumeID)
 	}
 	defer gceCS.volumeLocks.Release(volumeID)
+	disk, err := gceCS.createSingleDisk(ctx, req, params, volKey, zones)
+
+	if err != nil {
+		return nil, common.LoggedError("CreateVolume failed: %v", err)
+	}
+
+	return generateCreateVolumeResponseWithVolumeId(disk, zones, params, volumeID), err
+}
+
+func (gceCS *GCEControllerServer) createSingleDisk(ctx context.Context, req *csi.CreateVolumeRequest, params common.DiskParameters, volKey *meta.Key, zones []string) (*gce.CloudDisk, error) {
+	capacityRange := req.GetCapacityRange()
+	capBytes, _ := getRequestCapacity(capacityRange)
+	multiWriter, _ := getMultiWriterFromCapabilities(req.GetVolumeCapabilities())
+	readonly, _ := getReadOnlyFromCapabilities(req.GetVolumeCapabilities())
+	accessMode := ""
+	if readonly && slices.Contains(disksWithModifiableAccessMode, params.DiskType) {
+		accessMode = readOnlyManyAccessMode
+	}
 
 	// Validate if disk already exists
-	existingDisk, err := gceCS.CloudProvider.GetDisk(ctx, gceCS.CloudProvider.GetDefaultProject(), volKey, gceAPIVersion)
+	existingDisk, err := gceCS.CloudProvider.GetDisk(ctx, gceCS.CloudProvider.GetDefaultProject(), volKey, getGCEApiVersion(multiWriter))
 	if err != nil {
 		if !gce.IsGCEError(err, "notFound") {
 			// failed to GetDisk, however the Disk may already be created, the error code should be non-Final
@@ -409,7 +601,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 
 		// If there is no validation error, immediately return success
 		klog.V(4).Infof("CreateVolume succeeded for disk %v, it already exists and was compatible", volKey)
-		return generateCreateVolumeResponse(existingDisk, zones, params)
+		return existingDisk, nil
 	}
 
 	snapshotID := ""
@@ -437,7 +629,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 			}
 
 			// Verify that the volume in VolumeContentSource exists.
-			diskFromSourceVolume, err := gceCS.CloudProvider.GetDisk(ctx, project, sourceVolKey, gceAPIVersion)
+			diskFromSourceVolume, err := gceCS.CloudProvider.GetDisk(ctx, project, sourceVolKey, getGCEApiVersion(multiWriter))
 			if err != nil {
 				if gce.IsGCEError(err, "notFound") {
 					return nil, status.Errorf(codes.NotFound, "CreateVolume source volume %s does not exist", volumeContentSourceVolumeID)
@@ -485,20 +677,18 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 				return nil, status.Errorf(codes.Aborted, "CreateVolume disk from source volume %v is not ready", sourceVolKey)
 			}
 		}
-	} else { // if VolumeContentSource is nil, validate access mode is not read only
-		if readonly, _ := getReadOnlyFromCapabilities(volumeCapabilities); readonly {
-			return nil, status.Error(codes.InvalidArgument, "VolumeContentSource must be provided when AccessMode is set to read only")
-		}
 	}
 
 	// Create the disk
 	var disk *gce.CloudDisk
+	name := req.GetName()
+
 	switch params.ReplicationType {
 	case replicationTypeNone:
 		if len(zones) != 1 {
 			return nil, status.Errorf(codes.Internal, "CreateVolume failed to get a single zone for creating zonal disk, instead got: %v", zones)
 		}
-		disk, err = createSingleZoneDisk(ctx, gceCS.CloudProvider, name, zones, params, capacityRange, capBytes, snapshotID, volumeContentSourceVolumeID, multiWriter)
+		disk, err = createSingleZoneDisk(ctx, gceCS.CloudProvider, name, zones, params, capacityRange, capBytes, snapshotID, volumeContentSourceVolumeID, multiWriter, accessMode)
 		if err != nil {
 			return nil, common.LoggedError("CreateVolume failed to create single zonal disk "+name+": ", err)
 		}
@@ -506,7 +696,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 		if len(zones) != 2 {
 			return nil, status.Errorf(codes.Internal, "CreateVolume failed to get a 2 zones for creating regional disk, instead got: %v", zones)
 		}
-		disk, err = createRegionalDisk(ctx, gceCS.CloudProvider, name, zones, params, capacityRange, capBytes, snapshotID, volumeContentSourceVolumeID, multiWriter)
+		disk, err = createRegionalDisk(ctx, gceCS.CloudProvider, name, zones, params, capacityRange, capBytes, snapshotID, volumeContentSourceVolumeID, multiWriter, accessMode)
 		if err != nil {
 			return nil, common.LoggedError("CreateVolume failed to create regional disk "+name+": ", err)
 		}
@@ -523,18 +713,11 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 	}
 
 	klog.V(4).Infof("CreateVolume succeeded for disk %v", volKey)
-	return generateCreateVolumeResponse(disk, zones, params)
-
+	return disk, nil
 }
 
 func (gceCS *GCEControllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVolumeRequest) (*csi.DeleteVolumeResponse, error) {
 	var err error
-	diskTypeForMetric := metrics.DefaultDiskTypeForMetric
-	enableConfidentialCompute := metrics.DefaultEnableConfidentialCompute
-	enableStoragePools := metrics.DefaultEnableStoragePools
-	defer func() {
-		gceCS.Metrics.RecordOperationErrorMetrics("DeleteVolume", err, diskTypeForMetric, enableConfidentialCompute, enableStoragePools)
-	}()
 	// Validate arguments
 	volumeID := req.GetVolumeId()
 	if len(volumeID) == 0 {
@@ -549,6 +732,78 @@ func (gceCS *GCEControllerServer) DeleteVolume(ctx context.Context, req *csi.Del
 		return &csi.DeleteVolumeResponse{}, nil
 	}
 
+	volumeIsMultiZone := isMultiZoneVolKey(volKey)
+	if gceCS.multiZoneVolumeHandleConfig.Enable && volumeIsMultiZone {
+		// Delete multi-zone disk, that may have up to N disks.
+		return gceCS.deleteMultiZoneDisk(ctx, req, project, volKey)
+	}
+
+	// Delete zonal or regional disk
+	return gceCS.deleteSingleDeviceDisk(ctx, req, project, volKey)
+}
+
+func getGCEApiVersion(multiWriter bool) gce.GCEAPIVersion {
+	if multiWriter {
+		return gce.GCEAPIVersionBeta
+	}
+
+	return gce.GCEAPIVersionV1
+}
+
+func (gceCS *GCEControllerServer) deleteMultiZoneDisk(ctx context.Context, req *csi.DeleteVolumeRequest, project string, volKey *meta.Key) (*csi.DeleteVolumeResponse, error) {
+	// List disks with same name
+	var err error
+	diskTypeForMetric := metrics.DefaultDiskTypeForMetric
+	enableConfidentialCompute := metrics.DefaultEnableConfidentialCompute
+	enableStoragePools := metrics.DefaultEnableStoragePools
+	defer func() {
+		gceCS.Metrics.RecordOperationErrorMetrics("DeleteVolume", err, diskTypeForMetric, enableConfidentialCompute, enableStoragePools)
+	}()
+	existingZones := []string{gceCS.CloudProvider.GetDefaultZone()}
+	zones, err := getDefaultZonesInRegion(ctx, gceCS, existingZones)
+	if err != nil {
+		return nil, fmt.Errorf("failed to list default zones: %w", err)
+	}
+
+	volumeID := req.GetVolumeId()
+	if acquired := gceCS.volumeLocks.TryAcquire(volumeID); !acquired {
+		return nil, status.Errorf(codes.Aborted, common.VolumeOperationAlreadyExistsFmt, volumeID)
+	}
+	defer gceCS.volumeLocks.Release(volumeID)
+
+	deleteDiskErrs := []error{}
+	for _, zone := range zones {
+		zonalVolKey := &meta.Key{
+			Name:   volKey.Name,
+			Region: volKey.Region,
+			Zone:   zone,
+		}
+		disk, _ := gceCS.CloudProvider.GetDisk(ctx, project, zonalVolKey, gce.GCEAPIVersionV1)
+		// TODO: Consolidate the parameters here, rather than taking the last.
+		diskTypeForMetric, enableConfidentialCompute, enableStoragePools = metrics.GetMetricParameters(disk)
+		err := gceCS.CloudProvider.DeleteDisk(ctx, project, zonalVolKey)
+		if err != nil {
+			deleteDiskErrs = append(deleteDiskErrs, gceCS.CloudProvider.DeleteDisk(ctx, project, volKey))
+		}
+	}
+
+	if len(deleteDiskErrs) > 0 {
+		return nil, common.LoggedError("Failed to delete multi-zone disk: ", errors.Join(deleteDiskErrs...))
+	}
+
+	klog.V(4).Infof("DeleteVolume succeeded for disk %v", volKey)
+	return &csi.DeleteVolumeResponse{}, nil
+}
+
+func (gceCS *GCEControllerServer) deleteSingleDeviceDisk(ctx context.Context, req *csi.DeleteVolumeRequest, project string, volKey *meta.Key) (*csi.DeleteVolumeResponse, error) {
+	var err error
+	diskTypeForMetric := metrics.DefaultDiskTypeForMetric
+	enableConfidentialCompute := metrics.DefaultEnableConfidentialCompute
+	enableStoragePools := metrics.DefaultEnableStoragePools
+	defer func() {
+		gceCS.Metrics.RecordOperationErrorMetrics("DeleteVolume", err, diskTypeForMetric, enableConfidentialCompute, enableStoragePools)
+	}()
+	volumeID := req.GetVolumeId()
 	project, volKey, err = gceCS.CloudProvider.RepairUnderspecifiedVolumeKey(ctx, project, volKey)
 	if err != nil {
 		if gce.IsGCENotFoundError(err) {
@@ -661,6 +916,33 @@ func (gceCS *GCEControllerServer) validateMultiZoneDisk(volumeID string, disk *g
 	return nil
 }
 
+func (gceCS *GCEControllerServer) validateMultiZoneProvisioning(req *csi.CreateVolumeRequest, params common.DiskParameters) error {
+	if !gceCS.multiZoneVolumeHandleConfig.Enable {
+		return nil
+	}
+	if !params.MultiZoneProvisioning {
+		return nil
+	}
+
+	// For volume populator, we want to allow multiple RWO disks to be created
+	// with the same name, so they can be hydrated across multiple zones.
+
+	// We don't have support volume cloning from an existing PVC
+	if useVolumeCloning(req) {
+		return fmt.Errorf("%q parameter does not support volume cloning", common.ParameterKeyEnableMultiZoneProvisioning)
+	}
+
+	if readonly, _ := getReadOnlyFromCapabilities(req.GetVolumeCapabilities()); !readonly && req.GetVolumeContentSource() != nil {
+		return fmt.Errorf("%q parameter does not support specifying volume content source in readwrite mode", common.ParameterKeyEnableMultiZoneProvisioning)
+	}
+
+	if !slices.Contains(gceCS.multiZoneVolumeHandleConfig.DiskTypes, params.DiskType) {
+		return fmt.Errorf("%q parameter with unsupported disk type: %v", common.ParameterKeyEnableMultiZoneProvisioning, params.DiskType)
+	}
+
+	return nil
+}
+
 func isMultiZoneVolKey(volumeKey *meta.Key) bool {
 	return volumeKey.Type() == meta.Zonal && volumeKey.Zone == common.MultiZoneValue
 }
@@ -750,6 +1032,9 @@ func (gceCS *GCEControllerServer) executeControllerPublishVolume(ctx context.Con
 		klog.V(4).Infof("ControllerPublishVolume succeeded for disk %v to instance %v, already attached.", volKey, nodeID)
 		return pubVolResp, nil, disk
 	}
+	if err := gceCS.updateAccessModeIfNecessary(ctx, volKey, disk, readOnly); err != nil {
+		return nil, common.LoggedError("Failed to update access mode: ", err), disk
+	}
 	instanceZone, instanceName, err = common.NodeIDToZoneAndName(nodeID)
 	if err != nil {
 		return nil, status.Errorf(codes.InvalidArgument, "could not split nodeID: %v", err.Error()), disk
@@ -891,6 +1176,14 @@ func (gceCS *GCEControllerServer) executeControllerUnpublishVolume(ctx context.C
 	return &csi.ControllerUnpublishVolumeResponse{}, nil, diskToUnpublish
 }
 
+func (gceCS *GCEControllerServer) parameterProcessor() *common.ParameterProcessor {
+	return &common.ParameterProcessor{
+		DriverName:         gceCS.Driver.name,
+		EnableStoragePools: gceCS.enableStoragePools,
+		EnableMultiZone:    gceCS.multiZoneVolumeHandleConfig.Enable,
+	}
+}
+
 func (gceCS *GCEControllerServer) ValidateVolumeCapabilities(ctx context.Context, req *csi.ValidateVolumeCapabilitiesRequest) (*csi.ValidateVolumeCapabilitiesResponse, error) {
 	var err error
 	diskTypeForMetric := metrics.DefaultDiskTypeForMetric
@@ -939,7 +1232,7 @@ func (gceCS *GCEControllerServer) ValidateVolumeCapabilities(ctx context.Context
 	}
 
 	// Validate the disk parameters match the disk we GET
-	params, err := common.ExtractAndDefaultParameters(req.GetParameters(), gceCS.Driver.name, gceCS.Driver.extraVolumeLabels, gceCS.enableStoragePools, gceCS.Driver.extraTags)
+	params, err := gceCS.parameterProcessor().ExtractAndDefaultParameters(req.GetParameters(), gceCS.Driver.extraVolumeLabels, gceCS.Driver.extraTags)
 	if err != nil {
 		return nil, status.Errorf(codes.InvalidArgument, "failed to extract parameters: %v", err.Error())
 	}
@@ -1908,7 +2201,7 @@ func (gceCS *GCEControllerServer) pickZones(ctx context.Context, top *csi.Topolo
 			existingZones = []string{locationTopReq.srcVolZone}
 		}
 		// If topology is nil, then the Immediate binding mode was used without setting allowedTopologies in the storageclass.
-		zones, err = getDefaultZonesInRegion(ctx, gceCS, existingZones, numZones)
+		zones, err = getNumDefaultZonesInRegion(ctx, gceCS, existingZones, numZones)
 		if err != nil {
 			return nil, fmt.Errorf("failed to get default %v zones in region: %w", numZones, err)
 		}
@@ -1918,16 +2211,24 @@ func (gceCS *GCEControllerServer) pickZones(ctx context.Context, top *csi.Topolo
 	return zones, nil
 }
 
-func getDefaultZonesInRegion(ctx context.Context, gceCS *GCEControllerServer, existingZones []string, numZones int) ([]string, error) {
+func getDefaultZonesInRegion(ctx context.Context, gceCS *GCEControllerServer, existingZones []string) ([]string, error) {
 	region, err := common.GetRegionFromZones(existingZones)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get region from zones: %w", err)
 	}
-	needToGet := numZones - len(existingZones)
 	totZones, err := gceCS.CloudProvider.ListZones(ctx, region)
 	if err != nil {
 		return nil, fmt.Errorf("failed to list zones from cloud provider: %w", err)
 	}
+	return totZones, nil
+}
+
+func getNumDefaultZonesInRegion(ctx context.Context, gceCS *GCEControllerServer, existingZones []string, numZones int) ([]string, error) {
+	needToGet := numZones - len(existingZones)
+	totZones, err := getDefaultZonesInRegion(ctx, gceCS, existingZones)
+	if err != nil {
+		return nil, err
+	}
 	remainingZones := sets.NewString(totZones...).Difference(sets.NewString(existingZones...))
 	l := remainingZones.List()
 	if len(l) < needToGet {
@@ -1969,12 +2270,7 @@ func extractVolumeContext(context map[string]string) (*PDCSIContext, error) {
 	return info, nil
 }
 
-func generateCreateVolumeResponse(disk *gce.CloudDisk, zones []string, params common.DiskParameters) (*csi.CreateVolumeResponse, error) {
-	volumeId, err := getResourceId(disk.GetSelfLink())
-	if err != nil {
-		return nil, fmt.Errorf("cannot get volume id from %s: %w", disk.GetSelfLink(), err)
-	}
-
+func generateCreateVolumeResponseWithVolumeId(disk *gce.CloudDisk, zones []string, params common.DiskParameters, volumeId string) *csi.CreateVolumeResponse {
 	tops := []*csi.Topology{}
 	for _, zone := range zones {
 		tops = append(tops, &csi.Topology{
@@ -2024,7 +2320,7 @@ func generateCreateVolumeResponse(disk *gce.CloudDisk, zones []string, params co
 		}
 		createResp.Volume.ContentSource = contentSource
 	}
-	return createResp, nil
+	return createResp
 }
 
 func getResourceId(resourceLink string) (string, error) {
@@ -2056,7 +2352,7 @@ func getResourceId(resourceLink string) (string, error) {
 	return strings.Join(elts[3:], "/"), nil
 }
 
-func createRegionalDisk(ctx context.Context, cloudProvider gce.GCECompute, name string, zones []string, params common.DiskParameters, capacityRange *csi.CapacityRange, capBytes int64, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool) (*gce.CloudDisk, error) {
+func createRegionalDisk(ctx context.Context, cloudProvider gce.GCECompute, name string, zones []string, params common.DiskParameters, capacityRange *csi.CapacityRange, capBytes int64, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool, accessMode string) (*gce.CloudDisk, error) {
 	project := cloudProvider.GetDefaultProject()
 	region, err := common.GetRegionFromZones(zones)
 	if err != nil {
@@ -2069,7 +2365,7 @@ func createRegionalDisk(ctx context.Context, cloudProvider gce.GCECompute, name
 			fullyQualifiedReplicaZones, cloudProvider.GetReplicaZoneURI(project, replicaZone))
 	}
 
-	err = cloudProvider.InsertDisk(ctx, project, meta.RegionalKey(name, region), params, capBytes, capacityRange, fullyQualifiedReplicaZones, snapshotID, volumeContentSourceVolumeID, multiWriter)
+	err = cloudProvider.InsertDisk(ctx, project, meta.RegionalKey(name, region), params, capBytes, capacityRange, fullyQualifiedReplicaZones, snapshotID, volumeContentSourceVolumeID, multiWriter, accessMode)
 	if err != nil {
 		return nil, fmt.Errorf("failed to insert regional disk: %w", err)
 	}
@@ -2086,13 +2382,13 @@ func createRegionalDisk(ctx context.Context, cloudProvider gce.GCECompute, name
 	return disk, nil
 }
 
-func createSingleZoneDisk(ctx context.Context, cloudProvider gce.GCECompute, name string, zones []string, params common.DiskParameters, capacityRange *csi.CapacityRange, capBytes int64, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool) (*gce.CloudDisk, error) {
+func createSingleZoneDisk(ctx context.Context, cloudProvider gce.GCECompute, name string, zones []string, params common.DiskParameters, capacityRange *csi.CapacityRange, capBytes int64, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool, accessMode string) (*gce.CloudDisk, error) {
 	project := cloudProvider.GetDefaultProject()
 	if len(zones) != 1 {
 		return nil, fmt.Errorf("got wrong number of zones for zonal create volume: %v", len(zones))
 	}
 	diskZone := zones[0]
-	err := cloudProvider.InsertDisk(ctx, project, meta.ZonalKey(name, diskZone), params, capBytes, capacityRange, nil, snapshotID, volumeContentSourceVolumeID, multiWriter)
+	err := cloudProvider.InsertDisk(ctx, project, meta.ZonalKey(name, diskZone), params, capBytes, capacityRange, nil, snapshotID, volumeContentSourceVolumeID, multiWriter, accessMode)
 	if err != nil {
 		return nil, fmt.Errorf("failed to insert zonal disk: %w", err)
 	}
diff --git a/vendor/cloud.google.com/go/auth/CHANGES.md b/vendor/cloud.google.com/go/auth/CHANGES.md
index 7ef5fc0d..5a3f85ae 100644
--- a/vendor/cloud.google.com/go/auth/CHANGES.md
+++ b/vendor/cloud.google.com/go/auth/CHANGES.md
@@ -1,12 +1,5 @@
 # Changelog
 
-## [0.5.1](https://github.com/googleapis/google-cloud-go/compare/auth/v0.5.0...auth/v0.5.1) (2024-05-31)
-
-
-### Bug Fixes
-
-* **auth:** Pass through client to 2LO and 3LO flows ([#10290](https://github.com/googleapis/google-cloud-go/issues/10290)) ([685784e](https://github.com/googleapis/google-cloud-go/commit/685784ea84358c15e9214bdecb307d37aa3b6d2f))
-
 ## [0.5.0](https://github.com/googleapis/google-cloud-go/compare/auth/v0.4.2...auth/v0.5.0) (2024-05-28)
 
 
diff --git a/vendor/cloud.google.com/go/auth/credentials/filetypes.go b/vendor/cloud.google.com/go/auth/credentials/filetypes.go
index fe935573..a66e56d7 100644
--- a/vendor/cloud.google.com/go/auth/credentials/filetypes.go
+++ b/vendor/cloud.google.com/go/auth/credentials/filetypes.go
@@ -137,7 +137,6 @@ func handleServiceAccount(f *credsfile.ServiceAccountFile, opts *DetectOptions)
 		Scopes:       opts.scopes(),
 		TokenURL:     f.TokenURL,
 		Subject:      opts.Subject,
-		Client:       opts.client(),
 	}
 	if opts2LO.TokenURL == "" {
 		opts2LO.TokenURL = jwtTokenURL
@@ -155,7 +154,6 @@ func handleUserCredential(f *credsfile.UserCredentialsFile, opts *DetectOptions)
 		AuthStyle:        auth.StyleInParams,
 		EarlyTokenExpiry: opts.EarlyTokenRefresh,
 		RefreshToken:     f.RefreshToken,
-		Client:           opts.client(),
 	}
 	return auth.New3LOTokenProvider(opts3LO)
 }
diff --git a/vendor/cloud.google.com/go/auth/grpctransport/grpctransport.go b/vendor/cloud.google.com/go/auth/grpctransport/grpctransport.go
index 75bda4c6..81c956b0 100644
--- a/vendor/cloud.google.com/go/auth/grpctransport/grpctransport.go
+++ b/vendor/cloud.google.com/go/auth/grpctransport/grpctransport.go
@@ -47,7 +47,7 @@ var (
 
 // Options used to configure a [GRPCClientConnPool] from [Dial].
 type Options struct {
-	// DisableTelemetry disables default telemetry (OpenTelemetry). An example
+	// DisableTelemetry disables default telemetry (OpenCensus). An example
 	// reason to do so would be to bind custom telemetry that overrides the
 	// defaults.
 	DisableTelemetry bool
diff --git a/vendor/cloud.google.com/go/auth/httptransport/httptransport.go b/vendor/cloud.google.com/go/auth/httptransport/httptransport.go
index ef09c1b7..7fea9d87 100644
--- a/vendor/cloud.google.com/go/auth/httptransport/httptransport.go
+++ b/vendor/cloud.google.com/go/auth/httptransport/httptransport.go
@@ -33,7 +33,7 @@ type ClientCertProvider = func(*tls.CertificateRequestInfo) (*tls.Certificate, e
 
 // Options used to configure a [net/http.Client] from [NewClient].
 type Options struct {
-	// DisableTelemetry disables default telemetry (OpenTelemetry). An example
+	// DisableTelemetry disables default telemetry (OpenCensus). An example
 	// reason to do so would be to bind custom telemetry that overrides the
 	// defaults.
 	DisableTelemetry bool
diff --git a/vendor/golang.org/x/oauth2/google/google.go b/vendor/golang.org/x/oauth2/google/google.go
index 7b82e7a0..ba931c2c 100644
--- a/vendor/golang.org/x/oauth2/google/google.go
+++ b/vendor/golang.org/x/oauth2/google/google.go
@@ -252,10 +252,7 @@ func (f *credentialsFile) tokenSource(ctx context.Context, params CredentialsPar
 // Further information about retrieving access tokens from the GCE metadata
 // server can be found at https://cloud.google.com/compute/docs/authentication.
 func ComputeTokenSource(account string, scope ...string) oauth2.TokenSource {
-	// refresh 3 minutes and 45 seconds early. The shortest MDS cache is currently 4 minutes, so any
-	// refreshes earlier are a waste of compute.
-	earlyExpirySecs := 225 * time.Second
-	return computeTokenSource(account, earlyExpirySecs, scope...)
+	return computeTokenSource(account, 0, scope...)
 }
 
 func computeTokenSource(account string, earlyExpiry time.Duration, scope ...string) oauth2.TokenSource {
diff --git a/vendor/google.golang.org/api/compute/v0.alpha/compute-api.json b/vendor/google.golang.org/api/compute/v0.alpha/compute-api.json
index b4b3e56b..7d5c8416 100644
--- a/vendor/google.golang.org/api/compute/v0.alpha/compute-api.json
+++ b/vendor/google.golang.org/api/compute/v0.alpha/compute-api.json
@@ -19351,7 +19351,7 @@
           ]
         },
         "patch": {
-          "description": "Patches the specified network with the data included in the request. Only routingConfig can be modified.",
+          "description": "Patches the specified network with the data included in the request. Only the following fields can be modified: routingConfig.routingMode.",
           "flatPath": "projects/{project}/global/networks/{network}",
           "httpMethod": "PATCH",
           "id": "compute.networks.patch",
@@ -44423,7 +44423,7 @@
       }
     }
   },
-  "revision": "20240526",
+  "revision": "20240519",
   "rootUrl": "https://compute.googleapis.com/",
   "schemas": {
     "AWSV4Signature": {
@@ -46052,7 +46052,7 @@
           "description": "[Output Only] shielded vm initial state stored on disk"
         },
         "source": {
-          "description": "Specifies a valid partial or full URL to an existing Persistent Disk resource. When creating a new instance boot disk, one of initializeParams.sourceImage or initializeParams.sourceSnapshot or disks.source is required. If desired, you can also attach existing non-root persistent disks using this property. This field is only applicable for persistent disks. Note that for InstanceTemplate, specify the disk name for zonal disk, and the URL for regional disk.",
+          "description": "Specifies a valid partial or full URL to an existing Persistent Disk resource. When creating a new instance, one of initializeParams.sourceImage or initializeParams.sourceSnapshot or disks.source is required except for local SSD. If desired, you can also attach existing non-root persistent disks using this property. This field is only applicable for persistent disks. Note that for InstanceTemplate, specify the disk name for zonal disk, and the URL for regional disk.",
           "type": "string"
         },
         "type": {
@@ -46210,7 +46210,7 @@
           "type": "array"
         },
         "sourceImage": {
-          "description": "The source image to create this disk. When creating a new instance boot disk, one of initializeParams.sourceImage or initializeParams.sourceSnapshot or disks.source is required. To create a disk with one of the public operating system images, specify the image by its family name. For example, specify family/debian-9 to use the latest Debian 9 image: projects/debian-cloud/global/images/family/debian-9 Alternatively, use a specific version of a public operating system image: projects/debian-cloud/global/images/debian-9-stretch-vYYYYMMDD To create a disk with a custom image that you created, specify the image name in the following format: global/images/my-custom-image You can also specify a custom image by its image family, which returns the latest version of the image in that family. Replace the image name with family/family-name: global/images/family/my-image-family If the source image is deleted later, this field will not be set.",
+          "description": "The source image to create this disk. When creating a new instance, one of initializeParams.sourceImage or initializeParams.sourceSnapshot or disks.source is required except for local SSD. To create a disk with one of the public operating system images, specify the image by its family name. For example, specify family/debian-9 to use the latest Debian 9 image: projects/debian-cloud/global/images/family/debian-9 Alternatively, use a specific version of a public operating system image: projects/debian-cloud/global/images/debian-9-stretch-vYYYYMMDD To create a disk with a custom image that you created, specify the image name in the following format: global/images/my-custom-image You can also specify a custom image by its image family, which returns the latest version of the image in that family. Replace the image name with family/family-name: global/images/family/my-image-family If the source image is deleted later, this field will not be set.",
           "type": "string"
         },
         "sourceImageEncryptionKey": {
@@ -46218,11 +46218,11 @@
           "description": "The customer-supplied encryption key of the source image. Required if the source image is protected by a customer-supplied encryption key. InstanceTemplate and InstancePropertiesPatch do not store customer-supplied encryption keys, so you cannot create disks for instances in a managed instance group if the source images are encrypted with your own keys."
         },
         "sourceInstantSnapshot": {
-          "description": "The source instant-snapshot to create this disk. When creating a new instance boot disk, one of initializeParams.sourceSnapshot or initializeParams.sourceInstantSnapshot initializeParams.sourceImage or disks.source is required. To create a disk with a snapshot that you created, specify the snapshot name in the following format: us-central1-a/instantSnapshots/my-backup If the source instant-snapshot is deleted later, this field will not be set.",
+          "description": "The source instant-snapshot to create this disk. When creating a new instance, one of initializeParams.sourceSnapshot or initializeParams.sourceInstantSnapshot initializeParams.sourceImage or disks.source is required except for local SSD. To create a disk with a snapshot that you created, specify the snapshot name in the following format: us-central1-a/instantSnapshots/my-backup If the source instant-snapshot is deleted later, this field will not be set.",
           "type": "string"
         },
         "sourceSnapshot": {
-          "description": "The source snapshot to create this disk. When creating a new instance boot disk, one of initializeParams.sourceSnapshot or initializeParams.sourceImage or disks.source is required. To create a disk with a snapshot that you created, specify the snapshot name in the following format: global/snapshots/my-backup If the source snapshot is deleted later, this field will not be set.",
+          "description": "The source snapshot to create this disk. When creating a new instance, one of initializeParams.sourceSnapshot or initializeParams.sourceImage or disks.source is required except for local SSD. To create a disk with a snapshot that you created, specify the snapshot name in the following format: global/snapshots/my-backup If the source snapshot is deleted later, this field will not be set.",
           "type": "string"
         },
         "sourceSnapshotEncryptionKey": {
@@ -53589,22 +53589,6 @@
           },
           "type": "array"
         },
-        "destNetworkScope": {
-          "description": "Network scope of the traffic destination.",
-          "enum": [
-            "INTERNET",
-            "NON_INTERNET",
-            "UNSPECIFIED",
-            "VPC_NETWORKS"
-          ],
-          "enumDescriptions": [
-            "",
-            "",
-            "",
-            ""
-          ],
-          "type": "string"
-        },
         "destRegionCodes": {
           "description": "Region codes whose IP addresses will be used to match for destination of traffic. Should be specified as 2 letter country code defined as per ISO 3166 alpha-2 country codes. ex.\"US\" Maximum number of dest region codes allowed is 5000.",
           "items": {
@@ -53647,29 +53631,6 @@
           },
           "type": "array"
         },
-        "srcNetworkScope": {
-          "description": "Network scope of the traffic source.",
-          "enum": [
-            "INTERNET",
-            "NON_INTERNET",
-            "UNSPECIFIED",
-            "VPC_NETWORKS"
-          ],
-          "enumDescriptions": [
-            "",
-            "",
-            "",
-            ""
-          ],
-          "type": "string"
-        },
-        "srcNetworks": {
-          "description": "Networks of the traffic source. It can be either a full or partial url.",
-          "items": {
-            "type": "string"
-          },
-          "type": "array"
-        },
         "srcRegionCodes": {
           "description": "Region codes whose IP addresses will be used to match for source of traffic. Should be specified as 2 letter country code defined as per ISO 3166 alpha-2 country codes. ex.\"US\" Maximum number of source region codes allowed is 5000.",
           "items": {
@@ -59377,14 +59338,6 @@
           "description": "[Output Only] The URL of the region where the managed instance group resides (for regional resources).",
           "type": "string"
         },
-        "satisfiesPzi": {
-          "description": "[Output Only] Reserved for future use.",
-          "type": "boolean"
-        },
-        "satisfiesPzs": {
-          "description": "[Output Only] Reserved for future use.",
-          "type": "boolean"
-        },
         "selfLink": {
           "description": "[Output Only] The URL for this managed instance group. The server defines this URL.",
           "type": "string"
@@ -68460,10 +68413,6 @@
           "description": "A full or partial URL of the network placement to apply to this network. This field can be set only at resource creation time. For example, the following are valid URLs: - https://www.googleapis.com/compute/alpha/projects/{project_id}/global/networkPlacements/{network_placement_name} - projects/{project_id}/global/networkPlacements/{network_placement_name} ",
           "type": "string"
         },
-        "networkProfile": {
-          "description": "A full or partial URL of the network profile to apply to this network. This field can be set only at resource creation time. For example, the following are valid URLs: - https://www.googleapis.com/compute/alpha/projects/{project_id}/global/networkProfiles/{network_profile_name} - projects/{project_id}/global/networkProfiles/{network_profile_name} ",
-          "type": "string"
-        },
         "peerings": {
           "description": "[Output Only] A list of network peerings for the resource.",
           "items": {
@@ -99180,7 +99129,7 @@
       "type": "object"
     },
     "Zone": {
-      "description": "Represents a Zone resource. A zone is a deployment area. These deployment areas are subsets of a region. For example the zone us-east1-b is located in the us-east1 region. For more information, read Regions and Zones.",
+      "description": "Represents a Zone resource. A zone is a deployment area. These deployment areas are subsets of a region. For example the zone us-east1-a is located in the us-east1 region. For more information, read Regions and Zones.",
       "id": "Zone",
       "properties": {
         "availableCpuPlatforms": {
diff --git a/vendor/google.golang.org/api/compute/v0.alpha/compute-gen.go b/vendor/google.golang.org/api/compute/v0.alpha/compute-gen.go
index de1ce787..633d2049 100644
--- a/vendor/google.golang.org/api/compute/v0.alpha/compute-gen.go
+++ b/vendor/google.golang.org/api/compute/v0.alpha/compute-gen.go
@@ -3211,12 +3211,12 @@ type AttachedDisk struct {
 	// on disk
 	ShieldedInstanceInitialState *InitialStateConfig `json:"shieldedInstanceInitialState,omitempty"`
 	// Source: Specifies a valid partial or full URL to an existing Persistent Disk
-	// resource. When creating a new instance boot disk, one of
-	// initializeParams.sourceImage or initializeParams.sourceSnapshot or
-	// disks.source is required. If desired, you can also attach existing non-root
-	// persistent disks using this property. This field is only applicable for
-	// persistent disks. Note that for InstanceTemplate, specify the disk name for
-	// zonal disk, and the URL for regional disk.
+	// resource. When creating a new instance, one of initializeParams.sourceImage
+	// or initializeParams.sourceSnapshot or disks.source is required except for
+	// local SSD. If desired, you can also attach existing non-root persistent
+	// disks using this property. This field is only applicable for persistent
+	// disks. Note that for InstanceTemplate, specify the disk name for zonal disk,
+	// and the URL for regional disk.
 	Source string `json:"source,omitempty"`
 	// Type: Specifies the type of the disk, either SCRATCH or PERSISTENT. If not
 	// specified, the default is PERSISTENT.
@@ -3349,12 +3349,13 @@ type AttachedDiskInitializeParams struct {
 	// template, specify only the resource policy name.
 	ResourcePolicies []string `json:"resourcePolicies,omitempty"`
 	// SourceImage: The source image to create this disk. When creating a new
-	// instance boot disk, one of initializeParams.sourceImage or
-	// initializeParams.sourceSnapshot or disks.source is required. To create a
-	// disk with one of the public operating system images, specify the image by
-	// its family name. For example, specify family/debian-9 to use the latest
-	// Debian 9 image: projects/debian-cloud/global/images/family/debian-9
-	// Alternatively, use a specific version of a public operating system image:
+	// instance, one of initializeParams.sourceImage or
+	// initializeParams.sourceSnapshot or disks.source is required except for local
+	// SSD. To create a disk with one of the public operating system images,
+	// specify the image by its family name. For example, specify family/debian-9
+	// to use the latest Debian 9 image:
+	// projects/debian-cloud/global/images/family/debian-9 Alternatively, use a
+	// specific version of a public operating system image:
 	// projects/debian-cloud/global/images/debian-9-stretch-vYYYYMMDD To create a
 	// disk with a custom image that you created, specify the image name in the
 	// following format: global/images/my-custom-image You can also specify a
@@ -3371,19 +3372,19 @@ type AttachedDiskInitializeParams struct {
 	// keys.
 	SourceImageEncryptionKey *CustomerEncryptionKey `json:"sourceImageEncryptionKey,omitempty"`
 	// SourceInstantSnapshot: The source instant-snapshot to create this disk. When
-	// creating a new instance boot disk, one of initializeParams.sourceSnapshot or
+	// creating a new instance, one of initializeParams.sourceSnapshot or
 	// initializeParams.sourceInstantSnapshot initializeParams.sourceImage or
-	// disks.source is required. To create a disk with a snapshot that you created,
-	// specify the snapshot name in the following format:
-	// us-central1-a/instantSnapshots/my-backup If the source instant-snapshot is
-	// deleted later, this field will not be set.
+	// disks.source is required except for local SSD. To create a disk with a
+	// snapshot that you created, specify the snapshot name in the following
+	// format: us-central1-a/instantSnapshots/my-backup If the source
+	// instant-snapshot is deleted later, this field will not be set.
 	SourceInstantSnapshot string `json:"sourceInstantSnapshot,omitempty"`
 	// SourceSnapshot: The source snapshot to create this disk. When creating a new
-	// instance boot disk, one of initializeParams.sourceSnapshot or
-	// initializeParams.sourceImage or disks.source is required. To create a disk
-	// with a snapshot that you created, specify the snapshot name in the following
-	// format: global/snapshots/my-backup If the source snapshot is deleted later,
-	// this field will not be set.
+	// instance, one of initializeParams.sourceSnapshot or
+	// initializeParams.sourceImage or disks.source is required except for local
+	// SSD. To create a disk with a snapshot that you created, specify the snapshot
+	// name in the following format: global/snapshots/my-backup If the source
+	// snapshot is deleted later, this field will not be set.
 	SourceSnapshot string `json:"sourceSnapshot,omitempty"`
 	// SourceSnapshotEncryptionKey: The customer-supplied encryption key of the
 	// source snapshot.
@@ -12284,14 +12285,6 @@ type FirewallPolicyRuleMatcher struct {
 	// DestIpRanges: CIDR IP address range. Maximum number of destination CIDR IP
 	// ranges allowed is 5000.
 	DestIpRanges []string `json:"destIpRanges,omitempty"`
-	// DestNetworkScope: Network scope of the traffic destination.
-	//
-	// Possible values:
-	//   "INTERNET"
-	//   "NON_INTERNET"
-	//   "UNSPECIFIED"
-	//   "VPC_NETWORKS"
-	DestNetworkScope string `json:"destNetworkScope,omitempty"`
 	// DestRegionCodes: Region codes whose IP addresses will be used to match for
 	// destination of traffic. Should be specified as 2 letter country code defined
 	// as per ISO 3166 alpha-2 country codes. ex."US" Maximum number of dest region
@@ -12311,17 +12304,6 @@ type FirewallPolicyRuleMatcher struct {
 	// SrcIpRanges: CIDR IP address range. Maximum number of source CIDR IP ranges
 	// allowed is 5000.
 	SrcIpRanges []string `json:"srcIpRanges,omitempty"`
-	// SrcNetworkScope: Network scope of the traffic source.
-	//
-	// Possible values:
-	//   "INTERNET"
-	//   "NON_INTERNET"
-	//   "UNSPECIFIED"
-	//   "VPC_NETWORKS"
-	SrcNetworkScope string `json:"srcNetworkScope,omitempty"`
-	// SrcNetworks: Networks of the traffic source. It can be either a full or
-	// partial url.
-	SrcNetworks []string `json:"srcNetworks,omitempty"`
 	// SrcRegionCodes: Region codes whose IP addresses will be used to match for
 	// source of traffic. Should be specified as 2 letter country code defined as
 	// per ISO 3166 alpha-2 country codes. ex."US" Maximum number of source region
@@ -18819,10 +18801,6 @@ type InstanceGroupManager struct {
 	// Region: [Output Only] The URL of the region where the managed instance group
 	// resides (for regional resources).
 	Region string `json:"region,omitempty"`
-	// SatisfiesPzi: [Output Only] Reserved for future use.
-	SatisfiesPzi bool `json:"satisfiesPzi,omitempty"`
-	// SatisfiesPzs: [Output Only] Reserved for future use.
-	SatisfiesPzs bool `json:"satisfiesPzs,omitempty"`
 	// SelfLink: [Output Only] The URL for this managed instance group. The server
 	// defines this URL.
 	SelfLink string `json:"selfLink,omitempty"`
@@ -29264,12 +29242,6 @@ type Network struct {
 	// https://www.googleapis.com/compute/alpha/projects/{project_id}/global/networkPlacements/{network_placement_name}
 	// - projects/{project_id}/global/networkPlacements/{network_placement_name}
 	NetworkPlacement string `json:"networkPlacement,omitempty"`
-	// NetworkProfile: A full or partial URL of the network profile to apply to
-	// this network. This field can be set only at resource creation time. For
-	// example, the following are valid URLs: -
-	// https://www.googleapis.com/compute/alpha/projects/{project_id}/global/networkProfiles/{network_profile_name}
-	// - projects/{project_id}/global/networkProfiles/{network_profile_name}
-	NetworkProfile string `json:"networkProfile,omitempty"`
 	// Peerings: [Output Only] A list of network peerings for the resource.
 	Peerings []*NetworkPeering `json:"peerings,omitempty"`
 	// Region: [Output Only] URL of the region where the regional network resides.
@@ -62394,7 +62366,7 @@ func (s *XpnResourceId) MarshalJSON() ([]byte, error) {
 }
 
 // Zone: Represents a Zone resource. A zone is a deployment area. These
-// deployment areas are subsets of a region. For example the zone us-east1-b is
+// deployment areas are subsets of a region. For example the zone us-east1-a is
 // located in the us-east1 region. For more information, read Regions and
 // Zones.
 type Zone struct {
diff --git a/vendor/google.golang.org/api/compute/v0.alpha/compute2-gen.go b/vendor/google.golang.org/api/compute/v0.alpha/compute2-gen.go
index ba72997b..4f9d7a02 100644
--- a/vendor/google.golang.org/api/compute/v0.alpha/compute2-gen.go
+++ b/vendor/google.golang.org/api/compute/v0.alpha/compute2-gen.go
@@ -55939,7 +55939,7 @@ type NetworksPatchCall struct {
 }
 
 // Patch: Patches the specified network with the data included in the request.
-// Only routingConfig can be modified.
+// Only the following fields can be modified: routingConfig.routingMode.
 //
 // - network: Name of the network to update.
 // - project: Project ID for this request.
diff --git a/vendor/google.golang.org/api/compute/v0.beta/compute-api.json b/vendor/google.golang.org/api/compute/v0.beta/compute-api.json
index fc482286..5245aaab 100644
--- a/vendor/google.golang.org/api/compute/v0.beta/compute-api.json
+++ b/vendor/google.golang.org/api/compute/v0.beta/compute-api.json
@@ -18152,7 +18152,7 @@
           ]
         },
         "patch": {
-          "description": "Patches the specified network with the data included in the request. Only routingConfig can be modified.",
+          "description": "Patches the specified network with the data included in the request. Only the following fields can be modified: routingConfig.routingMode.",
           "flatPath": "projects/{project}/global/networks/{network}",
           "httpMethod": "PATCH",
           "id": "compute.networks.patch",
@@ -41579,7 +41579,7 @@
       }
     }
   },
-  "revision": "20240526",
+  "revision": "20240519",
   "rootUrl": "https://compute.googleapis.com/",
   "schemas": {
     "AWSV4Signature": {
@@ -43157,7 +43157,7 @@
           "description": "[Output Only] shielded vm initial state stored on disk"
         },
         "source": {
-          "description": "Specifies a valid partial or full URL to an existing Persistent Disk resource. When creating a new instance boot disk, one of initializeParams.sourceImage or initializeParams.sourceSnapshot or disks.source is required. If desired, you can also attach existing non-root persistent disks using this property. This field is only applicable for persistent disks. Note that for InstanceTemplate, specify the disk name for zonal disk, and the URL for regional disk.",
+          "description": "Specifies a valid partial or full URL to an existing Persistent Disk resource. When creating a new instance, one of initializeParams.sourceImage or initializeParams.sourceSnapshot or disks.source is required except for local SSD. If desired, you can also attach existing non-root persistent disks using this property. This field is only applicable for persistent disks. Note that for InstanceTemplate, specify the disk name for zonal disk, and the URL for regional disk.",
           "type": "string"
         },
         "type": {
@@ -43292,7 +43292,7 @@
           "type": "array"
         },
         "sourceImage": {
-          "description": "The source image to create this disk. When creating a new instance boot disk, one of initializeParams.sourceImage or initializeParams.sourceSnapshot or disks.source is required. To create a disk with one of the public operating system images, specify the image by its family name. For example, specify family/debian-9 to use the latest Debian 9 image: projects/debian-cloud/global/images/family/debian-9 Alternatively, use a specific version of a public operating system image: projects/debian-cloud/global/images/debian-9-stretch-vYYYYMMDD To create a disk with a custom image that you created, specify the image name in the following format: global/images/my-custom-image You can also specify a custom image by its image family, which returns the latest version of the image in that family. Replace the image name with family/family-name: global/images/family/my-image-family If the source image is deleted later, this field will not be set.",
+          "description": "The source image to create this disk. When creating a new instance, one of initializeParams.sourceImage or initializeParams.sourceSnapshot or disks.source is required except for local SSD. To create a disk with one of the public operating system images, specify the image by its family name. For example, specify family/debian-9 to use the latest Debian 9 image: projects/debian-cloud/global/images/family/debian-9 Alternatively, use a specific version of a public operating system image: projects/debian-cloud/global/images/debian-9-stretch-vYYYYMMDD To create a disk with a custom image that you created, specify the image name in the following format: global/images/my-custom-image You can also specify a custom image by its image family, which returns the latest version of the image in that family. Replace the image name with family/family-name: global/images/family/my-image-family If the source image is deleted later, this field will not be set.",
           "type": "string"
         },
         "sourceImageEncryptionKey": {
@@ -43300,11 +43300,11 @@
           "description": "The customer-supplied encryption key of the source image. Required if the source image is protected by a customer-supplied encryption key. InstanceTemplate and InstancePropertiesPatch do not store customer-supplied encryption keys, so you cannot create disks for instances in a managed instance group if the source images are encrypted with your own keys."
         },
         "sourceInstantSnapshot": {
-          "description": "The source instant-snapshot to create this disk. When creating a new instance boot disk, one of initializeParams.sourceSnapshot or initializeParams.sourceInstantSnapshot initializeParams.sourceImage or disks.source is required. To create a disk with a snapshot that you created, specify the snapshot name in the following format: us-central1-a/instantSnapshots/my-backup If the source instant-snapshot is deleted later, this field will not be set.",
+          "description": "The source instant-snapshot to create this disk. When creating a new instance, one of initializeParams.sourceSnapshot or initializeParams.sourceInstantSnapshot initializeParams.sourceImage or disks.source is required except for local SSD. To create a disk with a snapshot that you created, specify the snapshot name in the following format: us-central1-a/instantSnapshots/my-backup If the source instant-snapshot is deleted later, this field will not be set.",
           "type": "string"
         },
         "sourceSnapshot": {
-          "description": "The source snapshot to create this disk. When creating a new instance boot disk, one of initializeParams.sourceSnapshot or initializeParams.sourceImage or disks.source is required. To create a disk with a snapshot that you created, specify the snapshot name in the following format: global/snapshots/my-backup If the source snapshot is deleted later, this field will not be set.",
+          "description": "The source snapshot to create this disk. When creating a new instance, one of initializeParams.sourceSnapshot or initializeParams.sourceImage or disks.source is required except for local SSD. To create a disk with a snapshot that you created, specify the snapshot name in the following format: global/snapshots/my-backup If the source snapshot is deleted later, this field will not be set.",
           "type": "string"
         },
         "sourceSnapshotEncryptionKey": {
@@ -55415,14 +55415,6 @@
           "description": "[Output Only] The URL of the region where the managed instance group resides (for regional resources).",
           "type": "string"
         },
-        "satisfiesPzi": {
-          "description": "[Output Only] Reserved for future use.",
-          "type": "boolean"
-        },
-        "satisfiesPzs": {
-          "description": "[Output Only] Reserved for future use.",
-          "type": "boolean"
-        },
         "selfLink": {
           "description": "[Output Only] The URL for this managed instance group. The server defines this URL.",
           "type": "string"
@@ -55760,10 +55752,6 @@
           },
           "description": "Named instance selections configuring properties that the group will use when creating new VMs.",
           "type": "object"
-        },
-        "provisioningModelMix": {
-          "$ref": "InstanceGroupManagerInstanceFlexibilityPolicyProvisioningModelMix",
-          "description": "Provisioning model configuration used by this managed instance group to create instances."
         }
       },
       "type": "object"
@@ -55786,22 +55774,6 @@
       },
       "type": "object"
     },
-    "InstanceGroupManagerInstanceFlexibilityPolicyProvisioningModelMix": {
-      "id": "InstanceGroupManagerInstanceFlexibilityPolicyProvisioningModelMix",
-      "properties": {
-        "standardCapacityBase": {
-          "description": "The base capacity that will always use Standard VMs to avoid risk of more preemption than the minimum capacity user needs. MIG will create only Standard VMs until it reaches standard_capacity_base and only then will start using standard_capacity_percent_above_base to mix Spot with Standard VMs.",
-          "format": "int32",
-          "type": "integer"
-        },
-        "standardCapacityPercentAboveBase": {
-          "description": "The percentage of target capacity that should use Standard VM. The remaining percentage will use Spot VMs. The percentage applies only to the capacity above standard_capacity_base.",
-          "format": "int32",
-          "type": "integer"
-        }
-      },
-      "type": "object"
-    },
     "InstanceGroupManagerInstanceLifecyclePolicy": {
       "id": "InstanceGroupManagerInstanceLifecyclePolicy",
       "properties": {
@@ -63385,18 +63357,6 @@
         "machineType": {
           "description": "The machine type to be used for this instance.",
           "type": "string"
-        },
-        "provisioningModel": {
-          "description": "The provisioning model to be used for this instance.",
-          "enum": [
-            "SPOT",
-            "STANDARD"
-          ],
-          "enumDescriptions": [
-            "Heavily discounted, no guaranteed runtime.",
-            "Standard provisioning with user controlled runtime, no discounts."
-          ],
-          "type": "string"
         }
       },
       "type": "object"
@@ -91053,7 +91013,7 @@
       "type": "object"
     },
     "Zone": {
-      "description": "Represents a Zone resource. A zone is a deployment area. These deployment areas are subsets of a region. For example the zone us-east1-b is located in the us-east1 region. For more information, read Regions and Zones.",
+      "description": "Represents a Zone resource. A zone is a deployment area. These deployment areas are subsets of a region. For example the zone us-east1-a is located in the us-east1 region. For more information, read Regions and Zones.",
       "id": "Zone",
       "properties": {
         "availableCpuPlatforms": {
diff --git a/vendor/google.golang.org/api/compute/v0.beta/compute-gen.go b/vendor/google.golang.org/api/compute/v0.beta/compute-gen.go
index 75769462..67349491 100644
--- a/vendor/google.golang.org/api/compute/v0.beta/compute-gen.go
+++ b/vendor/google.golang.org/api/compute/v0.beta/compute-gen.go
@@ -3108,12 +3108,12 @@ type AttachedDisk struct {
 	// on disk
 	ShieldedInstanceInitialState *InitialStateConfig `json:"shieldedInstanceInitialState,omitempty"`
 	// Source: Specifies a valid partial or full URL to an existing Persistent Disk
-	// resource. When creating a new instance boot disk, one of
-	// initializeParams.sourceImage or initializeParams.sourceSnapshot or
-	// disks.source is required. If desired, you can also attach existing non-root
-	// persistent disks using this property. This field is only applicable for
-	// persistent disks. Note that for InstanceTemplate, specify the disk name for
-	// zonal disk, and the URL for regional disk.
+	// resource. When creating a new instance, one of initializeParams.sourceImage
+	// or initializeParams.sourceSnapshot or disks.source is required except for
+	// local SSD. If desired, you can also attach existing non-root persistent
+	// disks using this property. This field is only applicable for persistent
+	// disks. Note that for InstanceTemplate, specify the disk name for zonal disk,
+	// and the URL for regional disk.
 	Source string `json:"source,omitempty"`
 	// Type: Specifies the type of the disk, either SCRATCH or PERSISTENT. If not
 	// specified, the default is PERSISTENT.
@@ -3235,12 +3235,13 @@ type AttachedDiskInitializeParams struct {
 	// template, specify only the resource policy name.
 	ResourcePolicies []string `json:"resourcePolicies,omitempty"`
 	// SourceImage: The source image to create this disk. When creating a new
-	// instance boot disk, one of initializeParams.sourceImage or
-	// initializeParams.sourceSnapshot or disks.source is required. To create a
-	// disk with one of the public operating system images, specify the image by
-	// its family name. For example, specify family/debian-9 to use the latest
-	// Debian 9 image: projects/debian-cloud/global/images/family/debian-9
-	// Alternatively, use a specific version of a public operating system image:
+	// instance, one of initializeParams.sourceImage or
+	// initializeParams.sourceSnapshot or disks.source is required except for local
+	// SSD. To create a disk with one of the public operating system images,
+	// specify the image by its family name. For example, specify family/debian-9
+	// to use the latest Debian 9 image:
+	// projects/debian-cloud/global/images/family/debian-9 Alternatively, use a
+	// specific version of a public operating system image:
 	// projects/debian-cloud/global/images/debian-9-stretch-vYYYYMMDD To create a
 	// disk with a custom image that you created, specify the image name in the
 	// following format: global/images/my-custom-image You can also specify a
@@ -3257,19 +3258,19 @@ type AttachedDiskInitializeParams struct {
 	// keys.
 	SourceImageEncryptionKey *CustomerEncryptionKey `json:"sourceImageEncryptionKey,omitempty"`
 	// SourceInstantSnapshot: The source instant-snapshot to create this disk. When
-	// creating a new instance boot disk, one of initializeParams.sourceSnapshot or
+	// creating a new instance, one of initializeParams.sourceSnapshot or
 	// initializeParams.sourceInstantSnapshot initializeParams.sourceImage or
-	// disks.source is required. To create a disk with a snapshot that you created,
-	// specify the snapshot name in the following format:
-	// us-central1-a/instantSnapshots/my-backup If the source instant-snapshot is
-	// deleted later, this field will not be set.
+	// disks.source is required except for local SSD. To create a disk with a
+	// snapshot that you created, specify the snapshot name in the following
+	// format: us-central1-a/instantSnapshots/my-backup If the source
+	// instant-snapshot is deleted later, this field will not be set.
 	SourceInstantSnapshot string `json:"sourceInstantSnapshot,omitempty"`
 	// SourceSnapshot: The source snapshot to create this disk. When creating a new
-	// instance boot disk, one of initializeParams.sourceSnapshot or
-	// initializeParams.sourceImage or disks.source is required. To create a disk
-	// with a snapshot that you created, specify the snapshot name in the following
-	// format: global/snapshots/my-backup If the source snapshot is deleted later,
-	// this field will not be set.
+	// instance, one of initializeParams.sourceSnapshot or
+	// initializeParams.sourceImage or disks.source is required except for local
+	// SSD. To create a disk with a snapshot that you created, specify the snapshot
+	// name in the following format: global/snapshots/my-backup If the source
+	// snapshot is deleted later, this field will not be set.
 	SourceSnapshot string `json:"sourceSnapshot,omitempty"`
 	// SourceSnapshotEncryptionKey: The customer-supplied encryption key of the
 	// source snapshot.
@@ -17421,10 +17422,6 @@ type InstanceGroupManager struct {
 	// Region: [Output Only] The URL of the region where the managed instance group
 	// resides (for regional resources).
 	Region string `json:"region,omitempty"`
-	// SatisfiesPzi: [Output Only] Reserved for future use.
-	SatisfiesPzi bool `json:"satisfiesPzi,omitempty"`
-	// SatisfiesPzs: [Output Only] Reserved for future use.
-	SatisfiesPzs bool `json:"satisfiesPzs,omitempty"`
 	// SelfLink: [Output Only] The URL for this managed instance group. The server
 	// defines this URL.
 	SelfLink string `json:"selfLink,omitempty"`
@@ -17785,9 +17782,6 @@ type InstanceGroupManagerInstanceFlexibilityPolicy struct {
 	// InstanceSelections: Named instance selections configuring properties that
 	// the group will use when creating new VMs.
 	InstanceSelections map[string]InstanceGroupManagerInstanceFlexibilityPolicyInstanceSelection `json:"instanceSelections,omitempty"`
-	// ProvisioningModelMix: Provisioning model configuration used by this managed
-	// instance group to create instances.
-	ProvisioningModelMix *InstanceGroupManagerInstanceFlexibilityPolicyProvisioningModelMix `json:"provisioningModelMix,omitempty"`
 	// ForceSendFields is a list of field names (e.g. "InstanceSelectionLists") to
 	// unconditionally include in API requests. By default, fields with empty or
 	// default values are omitted from API requests. See
@@ -17832,35 +17826,6 @@ func (s *InstanceGroupManagerInstanceFlexibilityPolicyInstanceSelection) Marshal
 	return gensupport.MarshalJSON(NoMethod(*s), s.ForceSendFields, s.NullFields)
 }
 
-type InstanceGroupManagerInstanceFlexibilityPolicyProvisioningModelMix struct {
-	// StandardCapacityBase: The base capacity that will always use Standard VMs to
-	// avoid risk of more preemption than the minimum capacity user needs. MIG will
-	// create only Standard VMs until it reaches standard_capacity_base and only
-	// then will start using standard_capacity_percent_above_base to mix Spot with
-	// Standard VMs.
-	StandardCapacityBase int64 `json:"standardCapacityBase,omitempty"`
-	// StandardCapacityPercentAboveBase: The percentage of target capacity that
-	// should use Standard VM. The remaining percentage will use Spot VMs. The
-	// percentage applies only to the capacity above standard_capacity_base.
-	StandardCapacityPercentAboveBase int64 `json:"standardCapacityPercentAboveBase,omitempty"`
-	// ForceSendFields is a list of field names (e.g. "StandardCapacityBase") to
-	// unconditionally include in API requests. By default, fields with empty or
-	// default values are omitted from API requests. See
-	// https://pkg.go.dev/google.golang.org/api#hdr-ForceSendFields for more
-	// details.
-	ForceSendFields []string `json:"-"`
-	// NullFields is a list of field names (e.g. "StandardCapacityBase") to include
-	// in API requests with the JSON null value. By default, fields with empty
-	// values are omitted from API requests. See
-	// https://pkg.go.dev/google.golang.org/api#hdr-NullFields for more details.
-	NullFields []string `json:"-"`
-}
-
-func (s *InstanceGroupManagerInstanceFlexibilityPolicyProvisioningModelMix) MarshalJSON() ([]byte, error) {
-	type NoMethod InstanceGroupManagerInstanceFlexibilityPolicyProvisioningModelMix
-	return gensupport.MarshalJSON(NoMethod(*s), s.ForceSendFields, s.NullFields)
-}
-
 type InstanceGroupManagerInstanceLifecyclePolicy struct {
 	// DefaultActionOnFailure: The action that a MIG performs on a failed or an
 	// unhealthy VM. A VM is marked as unhealthy when the application running on
@@ -26606,13 +26571,6 @@ func (s *ManagedInstanceLastAttemptErrorsErrorsErrorDetails) MarshalJSON() ([]by
 type ManagedInstancePropertiesFromFlexibilityPolicy struct {
 	// MachineType: The machine type to be used for this instance.
 	MachineType string `json:"machineType,omitempty"`
-	// ProvisioningModel: The provisioning model to be used for this instance.
-	//
-	// Possible values:
-	//   "SPOT" - Heavily discounted, no guaranteed runtime.
-	//   "STANDARD" - Standard provisioning with user controlled runtime, no
-	// discounts.
-	ProvisioningModel string `json:"provisioningModel,omitempty"`
 	// ForceSendFields is a list of field names (e.g. "MachineType") to
 	// unconditionally include in API requests. By default, fields with empty or
 	// default values are omitted from API requests. See
@@ -56747,7 +56705,7 @@ func (s *XpnResourceId) MarshalJSON() ([]byte, error) {
 }
 
 // Zone: Represents a Zone resource. A zone is a deployment area. These
-// deployment areas are subsets of a region. For example the zone us-east1-b is
+// deployment areas are subsets of a region. For example the zone us-east1-a is
 // located in the us-east1 region. For more information, read Regions and
 // Zones.
 type Zone struct {
diff --git a/vendor/google.golang.org/api/compute/v0.beta/compute2-gen.go b/vendor/google.golang.org/api/compute/v0.beta/compute2-gen.go
index fb3ef16a..dee95b18 100644
--- a/vendor/google.golang.org/api/compute/v0.beta/compute2-gen.go
+++ b/vendor/google.golang.org/api/compute/v0.beta/compute2-gen.go
@@ -52511,7 +52511,7 @@ type NetworksPatchCall struct {
 }
 
 // Patch: Patches the specified network with the data included in the request.
-// Only routingConfig can be modified.
+// Only the following fields can be modified: routingConfig.routingMode.
 //
 // - network: Name of the network to update.
 // - project: Project ID for this request.
diff --git a/vendor/google.golang.org/api/compute/v1/compute-api.json b/vendor/google.golang.org/api/compute/v1/compute-api.json
index eecc2788..4a6b808b 100644
--- a/vendor/google.golang.org/api/compute/v1/compute-api.json
+++ b/vendor/google.golang.org/api/compute/v1/compute-api.json
@@ -16605,7 +16605,7 @@
           ]
         },
         "patch": {
-          "description": "Patches the specified network with the data included in the request. Only routingConfig can be modified.",
+          "description": "Patches the specified network with the data included in the request. Only the following fields can be modified: routingConfig.routingMode.",
           "flatPath": "projects/{project}/global/networks/{network}",
           "httpMethod": "PATCH",
           "id": "compute.networks.patch",
@@ -37421,7 +37421,7 @@
       }
     }
   },
-  "revision": "20240526",
+  "revision": "20240519",
   "rootUrl": "https://compute.googleapis.com/",
   "schemas": {
     "AWSV4Signature": {
@@ -38976,7 +38976,7 @@
           "description": "[Output Only] shielded vm initial state stored on disk"
         },
         "source": {
-          "description": "Specifies a valid partial or full URL to an existing Persistent Disk resource. When creating a new instance boot disk, one of initializeParams.sourceImage or initializeParams.sourceSnapshot or disks.source is required. If desired, you can also attach existing non-root persistent disks using this property. This field is only applicable for persistent disks. Note that for InstanceTemplate, specify the disk name for zonal disk, and the URL for regional disk.",
+          "description": "Specifies a valid partial or full URL to an existing Persistent Disk resource. When creating a new instance, one of initializeParams.sourceImage or initializeParams.sourceSnapshot or disks.source is required except for local SSD. If desired, you can also attach existing non-root persistent disks using this property. This field is only applicable for persistent disks. Note that for InstanceTemplate, specify the disk name for zonal disk, and the URL for regional disk.",
           "type": "string"
         },
         "type": {
@@ -39093,7 +39093,7 @@
           "type": "array"
         },
         "sourceImage": {
-          "description": "The source image to create this disk. When creating a new instance boot disk, one of initializeParams.sourceImage or initializeParams.sourceSnapshot or disks.source is required. To create a disk with one of the public operating system images, specify the image by its family name. For example, specify family/debian-9 to use the latest Debian 9 image: projects/debian-cloud/global/images/family/debian-9 Alternatively, use a specific version of a public operating system image: projects/debian-cloud/global/images/debian-9-stretch-vYYYYMMDD To create a disk with a custom image that you created, specify the image name in the following format: global/images/my-custom-image You can also specify a custom image by its image family, which returns the latest version of the image in that family. Replace the image name with family/family-name: global/images/family/my-image-family If the source image is deleted later, this field will not be set.",
+          "description": "The source image to create this disk. When creating a new instance, one of initializeParams.sourceImage or initializeParams.sourceSnapshot or disks.source is required except for local SSD. To create a disk with one of the public operating system images, specify the image by its family name. For example, specify family/debian-9 to use the latest Debian 9 image: projects/debian-cloud/global/images/family/debian-9 Alternatively, use a specific version of a public operating system image: projects/debian-cloud/global/images/debian-9-stretch-vYYYYMMDD To create a disk with a custom image that you created, specify the image name in the following format: global/images/my-custom-image You can also specify a custom image by its image family, which returns the latest version of the image in that family. Replace the image name with family/family-name: global/images/family/my-image-family If the source image is deleted later, this field will not be set.",
           "type": "string"
         },
         "sourceImageEncryptionKey": {
@@ -39101,7 +39101,7 @@
           "description": "The customer-supplied encryption key of the source image. Required if the source image is protected by a customer-supplied encryption key. InstanceTemplate and InstancePropertiesPatch do not store customer-supplied encryption keys, so you cannot create disks for instances in a managed instance group if the source images are encrypted with your own keys."
         },
         "sourceSnapshot": {
-          "description": "The source snapshot to create this disk. When creating a new instance boot disk, one of initializeParams.sourceSnapshot or initializeParams.sourceImage or disks.source is required. To create a disk with a snapshot that you created, specify the snapshot name in the following format: global/snapshots/my-backup If the source snapshot is deleted later, this field will not be set.",
+          "description": "The source snapshot to create this disk. When creating a new instance, one of initializeParams.sourceSnapshot or initializeParams.sourceImage or disks.source is required except for local SSD. To create a disk with a snapshot that you created, specify the snapshot name in the following format: global/snapshots/my-backup If the source snapshot is deleted later, this field will not be set.",
           "type": "string"
         },
         "sourceSnapshotEncryptionKey": {
@@ -50045,14 +50045,6 @@
           "description": "[Output Only] The URL of the region where the managed instance group resides (for regional resources).",
           "type": "string"
         },
-        "satisfiesPzi": {
-          "description": "[Output Only] Reserved for future use.",
-          "type": "boolean"
-        },
-        "satisfiesPzs": {
-          "description": "[Output Only] Reserved for future use.",
-          "type": "boolean"
-        },
         "selfLink": {
           "description": "[Output Only] The URL for this managed instance group. The server defines this URL.",
           "type": "string"
@@ -78330,20 +78322,6 @@
           "description": "URL of SslPolicy resource that will be associated with the TargetHttpsProxy resource. If not set, the TargetHttpsProxy resource has no SSL policy configured.",
           "type": "string"
         },
-        "tlsEarlyData": {
-          "description": " Specifies whether TLS 1.3 0-RTT Data (\"Early Data\") should be accepted for this service. Early Data allows a TLS resumption handshake to include the initial application payload (a HTTP request) alongside the handshake, reducing the effective round trips to \"zero\". This applies to TLS 1.3 connections over TCP (HTTP/2) as well as over UDP (QUIC/h3). This can improve application performance, especially on networks where interruptions may be common, such as on mobile. Requests with Early Data will have the \"Early-Data\" HTTP header set on the request, with a value of \"1\", to allow the backend to determine whether Early Data was included. Note: TLS Early Data may allow requests to be replayed, as the data is sent to the backend before the handshake has fully completed. Applications that allow idempotent HTTP methods to make non-idempotent changes, such as a GET request updating a database, should not accept Early Data on those requests, and reject requests with the \"Early-Data: 1\" HTTP header by returning a HTTP 425 (Too Early) status code, in order to remain RFC compliant. The default value is DISABLED.",
-          "enum": [
-            "DISABLED",
-            "PERMISSIVE",
-            "STRICT"
-          ],
-          "enumDescriptions": [
-            "TLS 1.3 Early Data is not advertised, and any (invalid) attempts to send Early Data will be rejected by closing the connection.",
-            "This enables TLS 1.3 0-RTT, and only allows Early Data to be included on requests with safe HTTP methods (GET, HEAD, OPTIONS, TRACE). This mode does not enforce any other limitations for requests with Early Data. The application owner should validate that Early Data is acceptable for a given request path.",
-            "This enables TLS 1.3 0-RTT, and only allows Early Data to be included on requests with safe HTTP methods (GET, HEAD, OPTIONS, TRACE) without query parameters. Requests that send Early Data with non-idempotent HTTP methods or with query parameters will be rejected with a HTTP 425."
-          ],
-          "type": "string"
-        },
         "urlMap": {
           "description": "A fully-qualified or valid partial URL to the UrlMap resource that defines the mapping from URL to the BackendService. For example, the following are all valid URLs for specifying a URL map: - https://www.googleapis.compute/v1/projects/project/global/urlMaps/ url-map - projects/project/global/urlMaps/url-map - global/urlMaps/url-map ",
           "type": "string"
@@ -83945,7 +83923,7 @@
       "type": "object"
     },
     "Zone": {
-      "description": "Represents a Zone resource. A zone is a deployment area. These deployment areas are subsets of a region. For example the zone us-east1-b is located in the us-east1 region. For more information, read Regions and Zones.",
+      "description": "Represents a Zone resource. A zone is a deployment area. These deployment areas are subsets of a region. For example the zone us-east1-a is located in the us-east1 region. For more information, read Regions and Zones.",
       "id": "Zone",
       "properties": {
         "availableCpuPlatforms": {
diff --git a/vendor/google.golang.org/api/compute/v1/compute-gen.go b/vendor/google.golang.org/api/compute/v1/compute-gen.go
index 2274778b..5e70e4d6 100644
--- a/vendor/google.golang.org/api/compute/v1/compute-gen.go
+++ b/vendor/google.golang.org/api/compute/v1/compute-gen.go
@@ -3053,12 +3053,12 @@ type AttachedDisk struct {
 	// on disk
 	ShieldedInstanceInitialState *InitialStateConfig `json:"shieldedInstanceInitialState,omitempty"`
 	// Source: Specifies a valid partial or full URL to an existing Persistent Disk
-	// resource. When creating a new instance boot disk, one of
-	// initializeParams.sourceImage or initializeParams.sourceSnapshot or
-	// disks.source is required. If desired, you can also attach existing non-root
-	// persistent disks using this property. This field is only applicable for
-	// persistent disks. Note that for InstanceTemplate, specify the disk name for
-	// zonal disk, and the URL for regional disk.
+	// resource. When creating a new instance, one of initializeParams.sourceImage
+	// or initializeParams.sourceSnapshot or disks.source is required except for
+	// local SSD. If desired, you can also attach existing non-root persistent
+	// disks using this property. This field is only applicable for persistent
+	// disks. Note that for InstanceTemplate, specify the disk name for zonal disk,
+	// and the URL for regional disk.
 	Source string `json:"source,omitempty"`
 	// Type: Specifies the type of the disk, either SCRATCH or PERSISTENT. If not
 	// specified, the default is PERSISTENT.
@@ -3168,12 +3168,13 @@ type AttachedDiskInitializeParams struct {
 	// template, specify only the resource policy name.
 	ResourcePolicies []string `json:"resourcePolicies,omitempty"`
 	// SourceImage: The source image to create this disk. When creating a new
-	// instance boot disk, one of initializeParams.sourceImage or
-	// initializeParams.sourceSnapshot or disks.source is required. To create a
-	// disk with one of the public operating system images, specify the image by
-	// its family name. For example, specify family/debian-9 to use the latest
-	// Debian 9 image: projects/debian-cloud/global/images/family/debian-9
-	// Alternatively, use a specific version of a public operating system image:
+	// instance, one of initializeParams.sourceImage or
+	// initializeParams.sourceSnapshot or disks.source is required except for local
+	// SSD. To create a disk with one of the public operating system images,
+	// specify the image by its family name. For example, specify family/debian-9
+	// to use the latest Debian 9 image:
+	// projects/debian-cloud/global/images/family/debian-9 Alternatively, use a
+	// specific version of a public operating system image:
 	// projects/debian-cloud/global/images/debian-9-stretch-vYYYYMMDD To create a
 	// disk with a custom image that you created, specify the image name in the
 	// following format: global/images/my-custom-image You can also specify a
@@ -3190,11 +3191,11 @@ type AttachedDiskInitializeParams struct {
 	// keys.
 	SourceImageEncryptionKey *CustomerEncryptionKey `json:"sourceImageEncryptionKey,omitempty"`
 	// SourceSnapshot: The source snapshot to create this disk. When creating a new
-	// instance boot disk, one of initializeParams.sourceSnapshot or
-	// initializeParams.sourceImage or disks.source is required. To create a disk
-	// with a snapshot that you created, specify the snapshot name in the following
-	// format: global/snapshots/my-backup If the source snapshot is deleted later,
-	// this field will not be set.
+	// instance, one of initializeParams.sourceSnapshot or
+	// initializeParams.sourceImage or disks.source is required except for local
+	// SSD. To create a disk with a snapshot that you created, specify the snapshot
+	// name in the following format: global/snapshots/my-backup If the source
+	// snapshot is deleted later, this field will not be set.
 	SourceSnapshot string `json:"sourceSnapshot,omitempty"`
 	// SourceSnapshotEncryptionKey: The customer-supplied encryption key of the
 	// source snapshot.
@@ -16031,10 +16032,6 @@ type InstanceGroupManager struct {
 	// Region: [Output Only] The URL of the region where the managed instance group
 	// resides (for regional resources).
 	Region string `json:"region,omitempty"`
-	// SatisfiesPzi: [Output Only] Reserved for future use.
-	SatisfiesPzi bool `json:"satisfiesPzi,omitempty"`
-	// SatisfiesPzs: [Output Only] Reserved for future use.
-	SatisfiesPzs bool `json:"satisfiesPzs,omitempty"`
 	// SelfLink: [Output Only] The URL for this managed instance group. The server
 	// defines this URL.
 	SelfLink string `json:"selfLink,omitempty"`
@@ -47037,36 +47034,6 @@ type TargetHttpsProxy struct {
 	// TargetHttpsProxy resource. If not set, the TargetHttpsProxy resource has no
 	// SSL policy configured.
 	SslPolicy string `json:"sslPolicy,omitempty"`
-	// TlsEarlyData:  Specifies whether TLS 1.3 0-RTT Data ("Early Data") should be
-	// accepted for this service. Early Data allows a TLS resumption handshake to
-	// include the initial application payload (a HTTP request) alongside the
-	// handshake, reducing the effective round trips to "zero". This applies to TLS
-	// 1.3 connections over TCP (HTTP/2) as well as over UDP (QUIC/h3). This can
-	// improve application performance, especially on networks where interruptions
-	// may be common, such as on mobile. Requests with Early Data will have the
-	// "Early-Data" HTTP header set on the request, with a value of "1", to allow
-	// the backend to determine whether Early Data was included. Note: TLS Early
-	// Data may allow requests to be replayed, as the data is sent to the backend
-	// before the handshake has fully completed. Applications that allow idempotent
-	// HTTP methods to make non-idempotent changes, such as a GET request updating
-	// a database, should not accept Early Data on those requests, and reject
-	// requests with the "Early-Data: 1" HTTP header by returning a HTTP 425 (Too
-	// Early) status code, in order to remain RFC compliant. The default value is
-	// DISABLED.
-	//
-	// Possible values:
-	//   "DISABLED" - TLS 1.3 Early Data is not advertised, and any (invalid)
-	// attempts to send Early Data will be rejected by closing the connection.
-	//   "PERMISSIVE" - This enables TLS 1.3 0-RTT, and only allows Early Data to
-	// be included on requests with safe HTTP methods (GET, HEAD, OPTIONS, TRACE).
-	// This mode does not enforce any other limitations for requests with Early
-	// Data. The application owner should validate that Early Data is acceptable
-	// for a given request path.
-	//   "STRICT" - This enables TLS 1.3 0-RTT, and only allows Early Data to be
-	// included on requests with safe HTTP methods (GET, HEAD, OPTIONS, TRACE)
-	// without query parameters. Requests that send Early Data with non-idempotent
-	// HTTP methods or with query parameters will be rejected with a HTTP 425.
-	TlsEarlyData string `json:"tlsEarlyData,omitempty"`
 	// UrlMap: A fully-qualified or valid partial URL to the UrlMap resource that
 	// defines the mapping from URL to the BackendService. For example, the
 	// following are all valid URLs for specifying a URL map: -
@@ -53211,7 +53178,7 @@ func (s *XpnResourceId) MarshalJSON() ([]byte, error) {
 }
 
 // Zone: Represents a Zone resource. A zone is a deployment area. These
-// deployment areas are subsets of a region. For example the zone us-east1-b is
+// deployment areas are subsets of a region. For example the zone us-east1-a is
 // located in the us-east1 region. For more information, read Regions and
 // Zones.
 type Zone struct {
diff --git a/vendor/google.golang.org/api/compute/v1/compute2-gen.go b/vendor/google.golang.org/api/compute/v1/compute2-gen.go
index 424c7e15..17970765 100644
--- a/vendor/google.golang.org/api/compute/v1/compute2-gen.go
+++ b/vendor/google.golang.org/api/compute/v1/compute2-gen.go
@@ -48414,7 +48414,7 @@ type NetworksPatchCall struct {
 }
 
 // Patch: Patches the specified network with the data included in the request.
-// Only routingConfig can be modified.
+// Only the following fields can be modified: routingConfig.routingMode.
 //
 // - network: Name of the network to update.
 // - project: Project ID for this request.
diff --git a/vendor/google.golang.org/api/internal/version.go b/vendor/google.golang.org/api/internal/version.go
index caf84413..66e8bfdb 100644
--- a/vendor/google.golang.org/api/internal/version.go
+++ b/vendor/google.golang.org/api/internal/version.go
@@ -5,4 +5,4 @@
 package internal
 
 // Version is the current tagged release of the library.
-const Version = "0.183.0"
+const Version = "0.182.0"
diff --git a/vendor/google.golang.org/genproto/googleapis/type/expr/expr.pb.go b/vendor/google.golang.org/genproto/googleapis/type/expr/expr.pb.go
index 7d57f34b..38ef56f7 100644
--- a/vendor/google.golang.org/genproto/googleapis/type/expr/expr.pb.go
+++ b/vendor/google.golang.org/genproto/googleapis/type/expr/expr.pb.go
@@ -1,4 +1,4 @@
-// Copyright 2024 Google LLC
+// Copyright 2021 Google LLC
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -15,7 +15,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
 // 	protoc-gen-go v1.26.0
-// 	protoc        v4.24.4
+// 	protoc        v3.12.2
 // source: google/type/expr.proto
 
 package expr
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 12702a37..a1793c71 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1,7 +1,7 @@
 # cloud.google.com/go v0.114.0
 ## explicit; go 1.20
 cloud.google.com/go
-# cloud.google.com/go/auth v0.5.1
+# cloud.google.com/go/auth v0.5.0
 ## explicit; go 1.20
 cloud.google.com/go/auth
 cloud.google.com/go/auth/credentials
@@ -438,7 +438,7 @@ golang.org/x/net/http2/hpack
 golang.org/x/net/idna
 golang.org/x/net/internal/timeseries
 golang.org/x/net/trace
-# golang.org/x/oauth2 v0.21.0
+# golang.org/x/oauth2 v0.20.0
 ## explicit; go 1.18
 golang.org/x/oauth2
 golang.org/x/oauth2/authhandler
@@ -512,7 +512,7 @@ golang.org/x/tools/internal/stdlib
 golang.org/x/tools/internal/tokeninternal
 golang.org/x/tools/internal/typesinternal
 golang.org/x/tools/internal/versions
-# google.golang.org/api v0.183.0
+# google.golang.org/api v0.182.0
 ## explicit; go 1.20
 google.golang.org/api/cloudresourcemanager/v1
 google.golang.org/api/compute/v0.alpha
@@ -531,18 +531,18 @@ google.golang.org/api/option/internaloption
 google.golang.org/api/transport/grpc
 google.golang.org/api/transport/http
 google.golang.org/api/transport/http/internal/propagation
-# google.golang.org/genproto v0.0.0-20240528184218-531527333157
-## explicit; go 1.20
+# google.golang.org/genproto v0.0.0-20240401170217-c3f982113cda
+## explicit; go 1.19
 google.golang.org/genproto/googleapis/cloud/kms/v1
 google.golang.org/genproto/googleapis/cloud/location
 google.golang.org/genproto/googleapis/type/expr
 google.golang.org/genproto/protobuf/field_mask
-# google.golang.org/genproto/googleapis/api v0.0.0-20240521202816-d264139d666e
-## explicit; go 1.20
+# google.golang.org/genproto/googleapis/api v0.0.0-20240513163218-0867130af1f8
+## explicit; go 1.19
 google.golang.org/genproto/googleapis/api
 google.golang.org/genproto/googleapis/api/annotations
 google.golang.org/genproto/googleapis/api/httpbody
-# google.golang.org/genproto/googleapis/rpc v0.0.0-20240528184218-531527333157
+# google.golang.org/genproto/googleapis/rpc v0.0.0-20240521202816-d264139d666e
 ## explicit; go 1.20
 google.golang.org/genproto/googleapis/rpc/code
 google.golang.org/genproto/googleapis/rpc/errdetails


Test patch:
diff --git a/pkg/common/parameters_test.go b/pkg/common/parameters_test.go
index b43aeaed..75f1c2a1 100644
--- a/pkg/common/parameters_test.go
+++ b/pkg/common/parameters_test.go
@@ -29,6 +29,7 @@ func TestExtractAndDefaultParameters(t *testing.T) {
 		parameters         map[string]string
 		labels             map[string]string
 		enableStoragePools bool
+		enableMultiZone    bool
 		extraTags          map[string]string
 		expectParams       DiskParameters
 		expectErr          bool
@@ -350,11 +351,53 @@ func TestExtractAndDefaultParameters(t *testing.T) {
 			labels:             map[string]string{},
 			expectErr:          true,
 		},
+		{
+			name:            "multi-zone-enable parameters, multi-zone label is set, multi-zone feature enabled",
+			parameters:      map[string]string{ParameterKeyType: "hyperdisk-ml", ParameterKeyEnableMultiZoneProvisioning: "true"},
+			labels:          map[string]string{MultiZoneLabel: "true"},
+			enableMultiZone: true,
+			expectParams: DiskParameters{
+				DiskType:              "hyperdisk-ml",
+				ReplicationType:       "none",
+				Tags:                  map[string]string{},
+				Labels:                map[string]string{MultiZoneLabel: "true"},
+				ResourceTags:          map[string]string{},
+				MultiZoneProvisioning: true,
+			},
+		},
+		{
+			name:            "multi-zone-enable parameters, multi-zone label is false, multi-zone feature enabled",
+			parameters:      map[string]string{ParameterKeyType: "hyperdisk-ml", ParameterKeyEnableMultiZoneProvisioning: "false"},
+			enableMultiZone: true,
+			expectParams: DiskParameters{
+				DiskType:        "hyperdisk-ml",
+				ReplicationType: "none",
+				Tags:            map[string]string{},
+				ResourceTags:    map[string]string{},
+				Labels:          map[string]string{},
+			},
+		},
+		{
+			name:            "multi-zone-enable parameters, invalid value, multi-zone feature enabled",
+			parameters:      map[string]string{ParameterKeyType: "hyperdisk-ml", ParameterKeyEnableMultiZoneProvisioning: "unknown"},
+			enableMultiZone: true,
+			expectErr:       true,
+		},
+		{
+			name:       "multi-zone-enable parameters, multi-zone label is set, multi-zone feature disabled",
+			parameters: map[string]string{ParameterKeyType: "hyperdisk-ml", ParameterKeyEnableMultiZoneProvisioning: "true"},
+			expectErr:  true,
+		},
 	}
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			p, err := ExtractAndDefaultParameters(tc.parameters, "testDriver", tc.labels, tc.enableStoragePools, tc.extraTags)
+			pp := ParameterProcessor{
+				DriverName:         "testDriver",
+				EnableStoragePools: tc.enableStoragePools,
+				EnableMultiZone:    tc.enableMultiZone,
+			}
+			p, err := pp.ExtractAndDefaultParameters(tc.parameters, tc.labels, tc.extraTags)
 			if gotErr := err != nil; gotErr != tc.expectErr {
 				t.Fatalf("ExtractAndDefaultParameters(%+v) = %v; expectedErr: %v", tc.parameters, err, tc.expectErr)
 			}
@@ -362,7 +405,7 @@ func TestExtractAndDefaultParameters(t *testing.T) {
 				return
 			}
 
-			if diff := cmp.Diff(p, tc.expectParams); diff != "" {
+			if diff := cmp.Diff(tc.expectParams, p); diff != "" {
 				t.Errorf("ExtractAndDefaultParameters(%+v): -want, +got \n%s", tc.parameters, diff)
 			}
 		})
diff --git a/pkg/common/utils_test.go b/pkg/common/utils_test.go
index e0c14f29..1b9b44e3 100644
--- a/pkg/common/utils_test.go
+++ b/pkg/common/utils_test.go
@@ -1648,3 +1648,89 @@ func TestUnorderedSlicesEqual(t *testing.T) {
 		})
 	}
 }
+
+func TestParseZoneFromURI(t *testing.T) {
+	testcases := []struct {
+		name      string
+		zoneURI   string
+		wantZone  string
+		expectErr bool
+	}{
+		{
+			name:     "ParseZoneFromURI_FullURI",
+			zoneURI:  "https://www.googleapis.com/compute/v1/projects/psch-gke-dev/zones/us-east4-a",
+			wantZone: "us-east4-a",
+		},
+		{
+			name:     "ParseZoneFromURI_ProjectZoneString",
+			zoneURI:  "projects/psch-gke-dev/zones/us-east4-a",
+			wantZone: "us-east4-a",
+		},
+		{
+			name:      "ParseZoneFromURI_Malformed",
+			zoneURI:   "projects/psch-gke-dev/regions/us-east4",
+			expectErr: true,
+		},
+	}
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			gotZone, err := ParseZoneFromURI(tc.zoneURI)
+			if err != nil && !tc.expectErr {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if err == nil && tc.expectErr {
+				t.Fatalf("Expected err, but none was returned. Zone result: %v", gotZone)
+			}
+			if gotZone != tc.wantZone {
+				t.Errorf("ParseZoneFromURI(%v): got %v, want %v", tc.zoneURI, gotZone, tc.wantZone)
+			}
+		})
+	}
+}
+
+func TestNewCombinedError(t *testing.T) {
+	testcases := []struct {
+		name     string
+		errors   []error
+		wantCode codes.Code
+	}{
+		{
+			name:     "single generic error",
+			errors:   []error{fmt.Errorf("my internal error")},
+			wantCode: codes.Internal,
+		},
+		{
+			name:     "single retryable error",
+			errors:   []error{&googleapi.Error{Code: http.StatusTooManyRequests, Message: "Resource Exhausted"}},
+			wantCode: codes.ResourceExhausted,
+		},
+		{
+			name:     "multi generic error",
+			errors:   []error{fmt.Errorf("my internal error"), fmt.Errorf("my other internal error")},
+			wantCode: codes.Internal,
+		},
+		{
+			name:     "multi retryable error",
+			errors:   []error{fmt.Errorf("my internal error"), &googleapi.Error{Code: http.StatusTooManyRequests, Message: "Resource Exhausted"}},
+			wantCode: codes.ResourceExhausted,
+		},
+		{
+			name:     "multi retryable error",
+			errors:   []error{fmt.Errorf("my internal error"), &googleapi.Error{Code: http.StatusGatewayTimeout, Message: "connection reset by peer"}, fmt.Errorf("my other internal error")},
+			wantCode: codes.Unavailable,
+		},
+		{
+			name:     "multi retryable error",
+			errors:   []error{fmt.Errorf("The disk resource is already being used"), &googleapi.Error{Code: http.StatusGatewayTimeout, Message: "connection reset by peer"}},
+			wantCode: codes.Unavailable,
+		},
+	}
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			gotCode := CodeForError(NewCombinedError("message", tc.errors))
+			if gotCode != tc.wantCode {
+				t.Errorf("NewCombinedError(%v): got %v, want %v", tc.errors, gotCode, tc.wantCode)
+			}
+		})
+	}
+}
diff --git a/pkg/gce-cloud-provider/compute/cloud-disk_test.go b/pkg/gce-cloud-provider/compute/cloud-disk_test.go
index d391212a..e2bea29d 100644
--- a/pkg/gce-cloud-provider/compute/cloud-disk_test.go
+++ b/pkg/gce-cloud-provider/compute/cloud-disk_test.go
@@ -148,3 +148,50 @@ func TestGetLabels(t *testing.T) {
 		}
 	}
 }
+
+func TestGetAccessMode(t *testing.T) {
+	testCases := []struct {
+		name           string
+		cloudDisk      *CloudDisk
+		wantAccessMode string
+	}{
+		{
+			name: "v1 disk accessMode",
+			cloudDisk: &CloudDisk{
+				disk: &computev1.Disk{
+					AccessMode: "READ_WRITE_SINGLE",
+				},
+			},
+			wantAccessMode: "READ_WRITE_SINGLE",
+		},
+		{
+			name: "beta disk accessMode",
+			cloudDisk: &CloudDisk{
+				betaDisk: &computebeta.Disk{
+					AccessMode: "READ_ONLY_MANY",
+				},
+			},
+			wantAccessMode: "READ_ONLY_MANY",
+		},
+		{
+			name: "unset disk accessMode",
+			cloudDisk: &CloudDisk{
+				betaDisk: &computebeta.Disk{},
+			},
+			wantAccessMode: "",
+		},
+		{
+			name:           "unset disk",
+			cloudDisk:      &CloudDisk{},
+			wantAccessMode: "",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Logf("Running test: %v", tc.name)
+		gotAccessMode := tc.cloudDisk.GetAccessMode()
+		if gotAccessMode != tc.wantAccessMode {
+			t.Errorf("GetAccessMode() got %v, want %v", gotAccessMode, tc.wantAccessMode)
+		}
+	}
+}
diff --git a/pkg/gce-cloud-provider/compute/gce_test.go b/pkg/gce-cloud-provider/compute/gce_test.go
index 49f85221..94bb4fd9 100644
--- a/pkg/gce-cloud-provider/compute/gce_test.go
+++ b/pkg/gce-cloud-provider/compute/gce_test.go
@@ -106,7 +106,7 @@ func TestGetComputeVersion(t *testing.T) {
 		name               string
 		computeEndpoint    *url.URL
 		computeEnvironment Environment
-		computeVersion     Version
+		computeVersion     GCEAPIVersion
 		expectedEndpoint   string
 		expectError        bool
 	}{
@@ -115,7 +115,7 @@ func TestGetComputeVersion(t *testing.T) {
 			name:               "check for production environment",
 			computeEndpoint:    convertStringToURL("https://compute.googleapis.com"),
 			computeEnvironment: EnvironmentProduction,
-			computeVersion:     versionBeta,
+			computeVersion:     GCEAPIVersionBeta,
 			expectedEndpoint:   "https://compute.googleapis.com/compute/beta/",
 			expectError:        false,
 		},
@@ -123,7 +123,7 @@ func TestGetComputeVersion(t *testing.T) {
 			name:               "check for staging environment",
 			computeEndpoint:    convertStringToURL("https://compute.googleapis.com"),
 			computeEnvironment: EnvironmentStaging,
-			computeVersion:     versionV1,
+			computeVersion:     GCEAPIVersionV1,
 			expectedEndpoint:   "https://compute.googleapis.com/compute/staging_v1/",
 			expectError:        false,
 		},
diff --git a/pkg/gce-pd-csi-driver/controller_test.go b/pkg/gce-pd-csi-driver/controller_test.go
index bee2e1d2..e196a1d3 100644
--- a/pkg/gce-pd-csi-driver/controller_test.go
+++ b/pkg/gce-pd-csi-driver/controller_test.go
@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 	"math/rand"
+	"net/http"
 	"reflect"
 	"sort"
 	"strconv"
@@ -30,11 +31,13 @@ import (
 	"google.golang.org/protobuf/types/known/timestamppb"
 
 	compute "google.golang.org/api/compute/v1"
+	"google.golang.org/api/googleapi"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"k8s.io/apimachinery/pkg/util/clock"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/util/flowcontrol"
+	"k8s.io/klog/v2"
 	"k8s.io/utils/strings/slices"
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
@@ -1131,6 +1134,583 @@ func TestCreateVolumeArguments(t *testing.T) {
 	}
 }
 
+func TestMultiZoneVolumeCreation(t *testing.T) {
+	testCases := []struct {
+		name               string
+		req                *csi.CreateVolumeRequest
+		enableStoragePools bool
+		fallbackZones      []string
+		expZones           []string
+		expErrCode         codes.Code
+	}{
+		{
+			name: "success single ROX multi-zone disk",
+			req: &csi.CreateVolumeRequest{
+				Name:          "test-name",
+				CapacityRange: stdCapRange,
+				VolumeCapabilities: []*csi.VolumeCapability{
+					{
+						AccessType: &csi.VolumeCapability_Mount{
+							Mount: &csi.VolumeCapability_MountVolume{},
+						},
+						AccessMode: &csi.VolumeCapability_AccessMode{
+							Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY,
+						},
+					},
+				},
+				Parameters: map[string]string{
+					common.ParameterKeyType:                        "hyperdisk-ml",
+					common.ParameterKeyEnableMultiZoneProvisioning: "true",
+				},
+				VolumeContentSource: &csi.VolumeContentSource{
+					Type: &csi.VolumeContentSource_Snapshot{
+						Snapshot: &csi.VolumeContentSource_SnapshotSource{
+							SnapshotId: testSnapshotID,
+						},
+					},
+				},
+				AccessibilityRequirements: &csi.TopologyRequirement{
+					Requisite: []*csi.Topology{
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+						},
+					},
+					Preferred: []*csi.Topology{
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+						},
+					},
+				},
+			},
+			expZones: []string{"us-central1-a"},
+		},
+		{
+			name: "single ROX multi-zone disk empty topology fallback zones",
+			req: &csi.CreateVolumeRequest{
+				Name:          "test-name",
+				CapacityRange: stdCapRange,
+				VolumeCapabilities: []*csi.VolumeCapability{
+					{
+						AccessType: &csi.VolumeCapability_Mount{
+							Mount: &csi.VolumeCapability_MountVolume{},
+						},
+						AccessMode: &csi.VolumeCapability_AccessMode{
+							Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY,
+						},
+					},
+				},
+				Parameters: map[string]string{
+					common.ParameterKeyType:                        "hyperdisk-ml",
+					common.ParameterKeyEnableMultiZoneProvisioning: "true",
+				},
+				VolumeContentSource: &csi.VolumeContentSource{
+					Type: &csi.VolumeContentSource_Snapshot{
+						Snapshot: &csi.VolumeContentSource_SnapshotSource{
+							SnapshotId: testSnapshotID,
+						},
+					},
+				},
+				AccessibilityRequirements: &csi.TopologyRequirement{},
+			},
+			fallbackZones: []string{zone, secondZone},
+			expZones:      []string{zone, secondZone},
+		},
+		{
+			name: "success triple ROX multi-zone disk",
+			req: &csi.CreateVolumeRequest{
+				Name:          "test-name",
+				CapacityRange: stdCapRange,
+				VolumeCapabilities: []*csi.VolumeCapability{
+					{
+						AccessType: &csi.VolumeCapability_Mount{
+							Mount: &csi.VolumeCapability_MountVolume{},
+						},
+						AccessMode: &csi.VolumeCapability_AccessMode{
+							Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY,
+						},
+					},
+				},
+				Parameters: map[string]string{
+					common.ParameterKeyType:                        "hyperdisk-ml",
+					common.ParameterKeyEnableMultiZoneProvisioning: "true",
+				},
+				VolumeContentSource: &csi.VolumeContentSource{
+					Type: &csi.VolumeContentSource_Snapshot{
+						Snapshot: &csi.VolumeContentSource_SnapshotSource{
+							SnapshotId: testSnapshotID,
+						},
+					},
+				},
+				AccessibilityRequirements: &csi.TopologyRequirement{
+					Requisite: []*csi.Topology{
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+						},
+					},
+					Preferred: []*csi.Topology{
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+						},
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-b"},
+						},
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-c"},
+						},
+					},
+				},
+			},
+			expZones: []string{"us-central1-a", "us-central1-b", "us-central1-c"},
+		},
+		{
+			name: "success triple rwo multi-zone disk",
+			req: &csi.CreateVolumeRequest{
+				Name:          "test-name",
+				CapacityRange: stdCapRange,
+				VolumeCapabilities: []*csi.VolumeCapability{
+					{
+						AccessType: &csi.VolumeCapability_Mount{
+							Mount: &csi.VolumeCapability_MountVolume{},
+						},
+						AccessMode: &csi.VolumeCapability_AccessMode{
+							Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+						},
+					},
+				},
+				Parameters: map[string]string{
+					common.ParameterKeyType:                        "hyperdisk-ml",
+					common.ParameterKeyEnableMultiZoneProvisioning: "true",
+				},
+				AccessibilityRequirements: &csi.TopologyRequirement{
+					Requisite: []*csi.Topology{
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+						},
+					},
+					Preferred: []*csi.Topology{
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+						},
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-b"},
+						},
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-c"},
+						},
+					},
+				},
+			},
+			expZones: []string{"us-central1-a", "us-central1-b", "us-central1-c"},
+		},
+		{
+			name: "err single ROX multi-zone no topology",
+			req: &csi.CreateVolumeRequest{
+				Name:          "test-name",
+				CapacityRange: stdCapRange,
+				VolumeCapabilities: []*csi.VolumeCapability{
+					{
+						AccessType: &csi.VolumeCapability_Mount{
+							Mount: &csi.VolumeCapability_MountVolume{},
+						},
+						AccessMode: &csi.VolumeCapability_AccessMode{
+							Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY,
+						},
+					},
+				},
+				Parameters: map[string]string{
+					common.ParameterKeyType:                        "hyperdisk-ml",
+					common.ParameterKeyEnableMultiZoneProvisioning: "true",
+				},
+				VolumeContentSource: &csi.VolumeContentSource{
+					Type: &csi.VolumeContentSource_Snapshot{
+						Snapshot: &csi.VolumeContentSource_SnapshotSource{
+							SnapshotId: testSnapshotID,
+						},
+					},
+				},
+			},
+			expErrCode: codes.InvalidArgument,
+		},
+		{
+			name: "err rwo access mode",
+			req: &csi.CreateVolumeRequest{
+				Name:          "test-name",
+				CapacityRange: stdCapRange,
+				VolumeCapabilities: []*csi.VolumeCapability{
+					{
+						AccessType: &csi.VolumeCapability_Mount{
+							Mount: &csi.VolumeCapability_MountVolume{},
+						},
+						AccessMode: &csi.VolumeCapability_AccessMode{
+							Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+						},
+					},
+				},
+				Parameters: map[string]string{
+					common.ParameterKeyType:                        "hyperdisk-ml",
+					common.ParameterKeyEnableMultiZoneProvisioning: "true",
+				},
+				VolumeContentSource: &csi.VolumeContentSource{
+					Type: &csi.VolumeContentSource_Snapshot{
+						Snapshot: &csi.VolumeContentSource_SnapshotSource{
+							SnapshotId: testSnapshotID,
+						},
+					},
+				},
+				AccessibilityRequirements: &csi.TopologyRequirement{
+					Requisite: []*csi.Topology{
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+						},
+					},
+					Preferred: []*csi.Topology{
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+						},
+					},
+				},
+			},
+			expErrCode: codes.InvalidArgument,
+		},
+		{
+			name: "err no content source",
+			req: &csi.CreateVolumeRequest{
+				Name:          "test-name",
+				CapacityRange: stdCapRange,
+				VolumeCapabilities: []*csi.VolumeCapability{
+					{
+						AccessType: &csi.VolumeCapability_Mount{
+							Mount: &csi.VolumeCapability_MountVolume{},
+						},
+						AccessMode: &csi.VolumeCapability_AccessMode{
+							Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY,
+						},
+					},
+				},
+				Parameters: map[string]string{
+					common.ParameterKeyType:                        "hyperdisk-ml",
+					common.ParameterKeyEnableMultiZoneProvisioning: "true",
+				},
+				AccessibilityRequirements: &csi.TopologyRequirement{
+					Requisite: []*csi.Topology{
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+						},
+					},
+					Preferred: []*csi.Topology{
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+						},
+					},
+				},
+			},
+			expErrCode: codes.InvalidArgument,
+		},
+		{
+			name: "err cloning not supported",
+			req: &csi.CreateVolumeRequest{
+				Name:          "test-name",
+				CapacityRange: stdCapRange,
+				VolumeCapabilities: []*csi.VolumeCapability{
+					{
+						AccessType: &csi.VolumeCapability_Mount{
+							Mount: &csi.VolumeCapability_MountVolume{},
+						},
+						AccessMode: &csi.VolumeCapability_AccessMode{
+							Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY,
+						},
+					},
+				},
+				VolumeContentSource: &csi.VolumeContentSource{
+					Type: &csi.VolumeContentSource_Volume{
+						Volume: &csi.VolumeContentSource_VolumeSource{
+							VolumeId: testVolumeID,
+						},
+					},
+				},
+				Parameters: map[string]string{
+					common.ParameterKeyType:                        "hyperdisk-ml",
+					common.ParameterKeyEnableMultiZoneProvisioning: "true",
+				},
+				AccessibilityRequirements: &csi.TopologyRequirement{
+					Requisite: []*csi.Topology{
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+						},
+					},
+					Preferred: []*csi.Topology{
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+						},
+					},
+				},
+			},
+			expErrCode: codes.InvalidArgument,
+		},
+	}
+
+	// Run test cases
+	for _, tc := range testCases {
+		t.Logf("test case: %s", tc.name)
+		// Setup new driver each time so no interference
+		fcp, err := gce.CreateFakeCloudProvider(project, zone, nil)
+		if err != nil {
+			t.Fatalf("Failed to create fake cloud provider: %v", err)
+		}
+		// Setup new driver each time so no interference
+		gceDriver := initGCEDriverWithCloudProvider(t, fcp)
+		gceDriver.cs.multiZoneVolumeHandleConfig.DiskTypes = []string{"hyperdisk-ml"}
+		gceDriver.cs.multiZoneVolumeHandleConfig.Enable = true
+		gceDriver.cs.fallbackRequisiteZones = tc.fallbackZones
+
+		if tc.req.VolumeContentSource.GetType() != nil {
+			snapshotParams, err := common.ExtractAndDefaultSnapshotParameters(nil, gceDriver.name, nil)
+			if err != nil {
+				t.Errorf("Got error extracting snapshot parameters: %v", err)
+			}
+			if snapshotParams.SnapshotType == common.DiskSnapshotType {
+				fcp.CreateSnapshot(context.Background(), project, meta.ZonalKey(name, common.MultiZoneValue), name, snapshotParams)
+			} else {
+				t.Fatalf("No volume source mentioned in snapshot parameters %v", snapshotParams)
+			}
+		}
+
+		// Start Test
+		resp, err := gceDriver.cs.CreateVolume(context.Background(), tc.req)
+		if err != nil {
+			serverError, ok := status.FromError(err)
+			if !ok {
+				t.Fatalf("Could not get error status code from err: %v", serverError)
+			}
+			if serverError.Code() != tc.expErrCode {
+				t.Fatalf("Expected error code: %v, got: %v. err : %v", tc.expErrCode, serverError.Code(), err)
+			}
+			continue
+		}
+		if tc.expErrCode != codes.OK {
+			t.Fatalf("Expected error: %v, got no error", tc.expErrCode)
+		}
+
+		topologies := make([]*csi.Topology, 0, len(tc.expZones))
+		for _, zone := range tc.expZones {
+			topologies = append(topologies, &csi.Topology{
+				Segments: map[string]string{common.TopologyKeyZone: zone},
+			})
+		}
+
+		expVol := &csi.Volume{
+			CapacityBytes:      common.GbToBytes(20),
+			VolumeId:           fmt.Sprintf("projects/%s/zones/multi-zone/disks/%s", project, name),
+			VolumeContext:      nil,
+			AccessibleTopology: topologies,
+			ContentSource:      tc.req.VolumeContentSource,
+		}
+
+		// Make sure responses match
+		vol := resp.GetVolume()
+		if vol == nil {
+			// If one is nil but not both
+			t.Fatalf("Expected volume %v, got nil volume", expVol)
+		}
+
+		klog.Warningf("Got accessible topology: %v", vol.GetAccessibleTopology())
+
+		sortTopologies := func(t1, t2 *csi.Topology) bool {
+			return t1.Segments[common.TopologyKeyZone] < t2.Segments[common.TopologyKeyZone]
+		}
+		if diff := cmp.Diff(expVol, vol, cmpopts.SortSlices(sortTopologies)); diff != "" {
+			t.Errorf("Accessible topologies mismatch (-want +got):\n%s", diff)
+		}
+
+		for _, zone := range tc.expZones {
+			volumeKey := meta.ZonalKey(name, zone)
+			disk, err := fcp.GetDisk(context.Background(), project, volumeKey, gce.GCEAPIVersionBeta)
+			if err != nil {
+				t.Fatalf("Get Disk failed for created disk with error: %v", err)
+			}
+			if disk.GetLabels()[common.MultiZoneLabel] != "true" {
+				t.Fatalf("Expect %s disk to have %s label, got: %v", volumeKey, common.MultiZoneLabel, disk.GetLabels())
+			}
+		}
+	}
+}
+
+type FakeCloudProviderInsertDiskErr struct {
+	*gce.FakeCloudProvider
+	insertDiskErrors map[string]error
+}
+
+func NewFakeCloudProviderInsertDiskErr(project, zone string) (*FakeCloudProviderInsertDiskErr, error) {
+	provider, err := gce.CreateFakeCloudProvider(project, zone, nil)
+	if err != nil {
+		return nil, err
+	}
+	return &FakeCloudProviderInsertDiskErr{
+		FakeCloudProvider: provider,
+		insertDiskErrors:  map[string]error{},
+	}, nil
+}
+
+func (cloud *FakeCloudProviderInsertDiskErr) AddDiskForErr(volKey *meta.Key, err error) {
+	cloud.insertDiskErrors[volKey.String()] = err
+}
+
+func (cloud *FakeCloudProviderInsertDiskErr) InsertDisk(ctx context.Context, project string, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool, accessMode string) error {
+	if err, ok := cloud.insertDiskErrors[volKey.String()]; ok {
+		return err
+	}
+
+	return cloud.FakeCloudProvider.InsertDisk(ctx, project, volKey, params, capBytes, capacityRange, replicaZones, snapshotID, volumeContentSourceVolumeID, multiWriter, accessMode)
+}
+
+func TestMultiZoneVolumeCreationErrHandling(t *testing.T) {
+	testCases := []struct {
+		name           string
+		req            *csi.CreateVolumeRequest
+		insertDiskErrs map[*meta.Key]error
+		expErrCode     codes.Code
+		wantDisks      []*meta.Key
+	}{
+		{
+			name: "ResourceExhausted errors",
+			req: &csi.CreateVolumeRequest{
+				Name:          "test-name",
+				CapacityRange: stdCapRange,
+				VolumeCapabilities: []*csi.VolumeCapability{
+					{
+						AccessType: &csi.VolumeCapability_Mount{
+							Mount: &csi.VolumeCapability_MountVolume{},
+						},
+						AccessMode: &csi.VolumeCapability_AccessMode{
+							Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+						},
+					},
+				},
+				Parameters: map[string]string{
+					common.ParameterKeyType:                        "hyperdisk-ml",
+					common.ParameterKeyEnableMultiZoneProvisioning: "true",
+				},
+				AccessibilityRequirements: &csi.TopologyRequirement{
+					Requisite: []*csi.Topology{
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+						},
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-b"},
+						},
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-c"},
+						},
+					},
+					Preferred: []*csi.Topology{
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+						},
+					},
+				},
+			},
+			insertDiskErrs: map[*meta.Key]error{
+				meta.ZonalKey(name, "us-central1-b"): &googleapi.Error{Code: http.StatusTooManyRequests, Message: "Resource Exhausted"},
+			},
+			expErrCode: codes.ResourceExhausted,
+			wantDisks: []*meta.Key{
+				meta.ZonalKey(name, "us-central1-a"),
+				meta.ZonalKey(name, "us-central1-c"),
+			},
+		},
+		{
+			name: "Unavailable errors",
+			req: &csi.CreateVolumeRequest{
+				Name:          "test-name",
+				CapacityRange: stdCapRange,
+				VolumeCapabilities: []*csi.VolumeCapability{
+					{
+						AccessType: &csi.VolumeCapability_Mount{
+							Mount: &csi.VolumeCapability_MountVolume{},
+						},
+						AccessMode: &csi.VolumeCapability_AccessMode{
+							Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+						},
+					},
+				},
+				Parameters: map[string]string{
+					common.ParameterKeyType:                        "hyperdisk-ml",
+					common.ParameterKeyEnableMultiZoneProvisioning: "true",
+				},
+				AccessibilityRequirements: &csi.TopologyRequirement{
+					Requisite: []*csi.Topology{
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+						},
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-b"},
+						},
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-c"},
+						},
+					},
+					Preferred: []*csi.Topology{
+						{
+							Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+						},
+					},
+				},
+			},
+			insertDiskErrs: map[*meta.Key]error{
+				meta.ZonalKey(name, "us-central1-b"): &googleapi.Error{Code: http.StatusGatewayTimeout, Message: "connection reset by peer"},
+				meta.ZonalKey(name, "us-central1-c"): &googleapi.Error{Code: http.StatusTooManyRequests, Message: "Resource Exhausted"},
+			},
+			expErrCode: codes.Unavailable,
+			wantDisks: []*meta.Key{
+				meta.ZonalKey(name, "us-central1-a"),
+			},
+		},
+	}
+
+	// Run test cases
+	for _, tc := range testCases {
+		t.Logf("test case: %s", tc.name)
+		// Setup new driver each time so no interference
+		fcp, err := NewFakeCloudProviderInsertDiskErr(project, zone)
+		if err != nil {
+			t.Fatalf("Failed to create fake cloud provider: %v", err)
+		}
+		// Setup new driver each time so no interference
+		gceDriver := initGCEDriverWithCloudProvider(t, fcp)
+		gceDriver.cs.multiZoneVolumeHandleConfig.DiskTypes = []string{"hyperdisk-ml"}
+		gceDriver.cs.multiZoneVolumeHandleConfig.Enable = true
+
+		for volKey, err := range tc.insertDiskErrs {
+			fcp.AddDiskForErr(volKey, err)
+		}
+
+		// Start Test
+		_, err = gceDriver.cs.CreateVolume(context.Background(), tc.req)
+
+		if err == nil {
+			t.Errorf("Expected error: %v, got no error", tc.expErrCode)
+		}
+
+		serverError, ok := status.FromError(err)
+		if !ok {
+			t.Errorf("Could not get error status code from err: %v", serverError)
+		}
+		if serverError.Code() != tc.expErrCode {
+			t.Errorf("Expected error code: %v, got: %v. err : %v", tc.expErrCode, serverError.Code(), err)
+		}
+
+		for _, volKey := range tc.wantDisks {
+			disk, err := fcp.GetDisk(context.Background(), project, volKey, gce.GCEAPIVersionV1)
+			if err != nil {
+				t.Errorf("Unexpected err fetching disk %v: %v", volKey, err)
+			}
+			if disk == nil {
+				t.Errorf("Expected disk for %v but got nil", volKey)
+			}
+		}
+	}
+}
+
 func TestListVolumePagination(t *testing.T) {
 	testCases := []struct {
 		name            string
@@ -2463,6 +3043,14 @@ func createZonalCloudDisk(name string) *gce.CloudDisk {
 	})
 }
 
+func createZonalCloudDiskWithZone(name, zone string) *gce.CloudDisk {
+	return gce.CloudDiskFromV1(&compute.Disk{
+		Name:     name,
+		SelfLink: fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/project/zones/zone/name/%s", name),
+		Zone:     zone,
+	})
+}
+
 func TestDeleteVolume(t *testing.T) {
 	testCases := []struct {
 		name      string
@@ -2527,6 +3115,63 @@ func TestDeleteVolume(t *testing.T) {
 	}
 }
 
+func TestMultiZoneDeleteVolume(t *testing.T) {
+	testCases := []struct {
+		name      string
+		seedDisks []*gce.CloudDisk
+		req       *csi.DeleteVolumeRequest
+		expErr    bool
+	}{
+		{
+			name: "single-zone",
+			seedDisks: []*gce.CloudDisk{
+				createZonalCloudDiskWithZone(name, zone),
+			},
+			req: &csi.DeleteVolumeRequest{
+				VolumeId: fmt.Sprintf("projects/%s/zones/%s/disks/%s", project, common.MultiZoneValue, name),
+			},
+		},
+		{
+			name: "multi-zone",
+			seedDisks: []*gce.CloudDisk{
+				createZonalCloudDiskWithZone(name, zone),
+				createZonalCloudDiskWithZone(name, secondZone),
+			},
+			req: &csi.DeleteVolumeRequest{
+				VolumeId: fmt.Sprintf("projects/%s/zones/%s/disks/%s", project, common.MultiZoneValue, name),
+			},
+		},
+	}
+	for _, tc := range testCases {
+		t.Logf("test case: %s", tc.name)
+		// Setup new driver each time so no interference
+		fcp, err := gce.CreateFakeCloudProvider(project, zone, tc.seedDisks)
+		if err != nil {
+			t.Fatalf("Failed to create fake cloud provider: %v", err)
+		}
+		// Setup new driver each time so no interference
+		gceDriver := initGCEDriverWithCloudProvider(t, fcp)
+		gceDriver.cs.multiZoneVolumeHandleConfig.DiskTypes = []string{"hyperdisk-ml"}
+		gceDriver.cs.multiZoneVolumeHandleConfig.Enable = true
+		_, err = gceDriver.cs.DeleteVolume(context.Background(), tc.req)
+		if err == nil && tc.expErr {
+			t.Errorf("Expected error but got none")
+		}
+		if err != nil && !tc.expErr {
+			t.Errorf("Did not expect error but got: %v", err)
+		}
+
+		disks, _, _ := fcp.ListDisks(context.TODO(), []googleapi.Field{})
+		if len(disks) > 0 {
+			t.Errorf("Expected all disks to be deleted. Got: %v", disks)
+		}
+
+		if err != nil {
+			continue
+		}
+	}
+}
+
 func TestGetRequestCapacity(t *testing.T) {
 	testCases := []struct {
 		name     string
diff --git a/test/e2e/tests/multi_zone_e2e_test.go b/test/e2e/tests/multi_zone_e2e_test.go
index 81ee01c0..fb1ea32d 100644
--- a/test/e2e/tests/multi_zone_e2e_test.go
+++ b/test/e2e/tests/multi_zone_e2e_test.go
@@ -16,8 +16,11 @@ package tests
 
 import (
 	"fmt"
+	"os"
 	"path/filepath"
+	"strconv"
 	"strings"
+	"time"
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	. "github.com/onsi/ginkgo/v2"
@@ -25,6 +28,7 @@ import (
 	compute "google.golang.org/api/compute/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/uuid"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/klog/v2"
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
 	gce "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute"
@@ -40,6 +44,28 @@ type verifyFunc func(*verifyArgs) error
 
 type detacherFunc func()
 
+func runMultiZoneTests() bool {
+	runMultiZoneTestsStr, ok := os.LookupEnv("RUN_MULTI_ZONE_TESTS")
+	if !ok {
+		return false
+	}
+
+	runMultiZoneTests, err := strconv.ParseBool(runMultiZoneTestsStr)
+	if err != nil {
+		return false
+	}
+
+	return runMultiZoneTests
+}
+
+func checkSkipMultiZoneTests() {
+	// TODO: Remove this once hyperdisk-ml SKU is supported
+	// If you want to run these tests, set the env variable: RUN_MULTI_ZONE_TESTS=true
+	if !runMultiZoneTests() {
+		Skip("Not running multi-zone tests, as RUN_MULTI_ZONE_TESTS is falsy")
+	}
+}
+
 var _ = Describe("GCE PD CSI Driver Multi-Zone", func() {
 	BeforeEach(func() {
 		Expect(len(testContexts)).To(BeNumerically(">", 1))
@@ -67,6 +93,8 @@ var _ = Describe("GCE PD CSI Driver Multi-Zone", func() {
 	})
 
 	It("Should attach ROX 'multi-zone' PV instances to two separate VMs", func() {
+		checkSkipMultiZoneTests()
+
 		// Create new driver and client
 
 		Expect(testContexts).NotTo(BeEmpty())
@@ -96,8 +124,8 @@ var _ = Describe("GCE PD CSI Driver Multi-Zone", func() {
 
 		// Create Disk
 		volName := testNamePrefix + string(uuid.NewUUID())
-		_, volID0 := createAndValidateZonalDisk(controllerClient, p, zones[0], standardDiskType, volName)
-		_, volID1 := createAndValidateZonalDisk(controllerClient, p, zones[1], standardDiskType, volName)
+		_, volID0 := createAndValidateZonalDisk(controllerClient, p, zones[0], "hyperdisk-ml", volName)
+		_, volID1 := createAndValidateZonalDisk(controllerClient, p, zones[1], "hyperdisk-ml", volName)
 
 		labelsMap := map[string]string{
 			common.MultiZoneLabel: "true",
@@ -156,6 +184,650 @@ var _ = Describe("GCE PD CSI Driver Multi-Zone", func() {
 
 		err = controllerClient.ControllerUnpublishVolume(volID, nodeID1)
 		Expect(err).To(BeNil(), "Failed to detach vol2")
+	})
+
+	It("Should create RWO 'multi-zone' PV instances from a previously created disk", func() {
+		checkSkipMultiZoneTests()
+
+		// Create new driver and client
+
+		Expect(testContexts).NotTo(BeEmpty())
+
+		zoneToContext := map[string]*remote.TestContext{}
+		zones := []string{}
+
+		for _, tc := range testContexts {
+			_, z, _ := tc.Instance.GetIdentity()
+			// Zone hasn't been seen before
+			if _, ok := zoneToContext[z]; !ok {
+				zoneToContext[z] = tc
+				zones = append(zones, z)
+			}
+			if len(zoneToContext) == 2 {
+				break
+			}
+		}
+
+		Expect(len(zoneToContext)).To(Equal(2), "Must have instances in 2 zones")
+
+		controllerContext := zoneToContext[zones[0]]
+		controllerClient := controllerContext.Client
+		controllerInstance := controllerContext.Instance
+
+		p, _, _ := controllerInstance.GetIdentity()
+
+		// Create Disk
+		volName := testNamePrefix + string(uuid.NewUUID())
+		_, volID0 := createAndValidateZonalDisk(controllerClient, p, zones[0], "hyperdisk-ml", volName)
+
+		labelsMap := map[string]string{
+			common.MultiZoneLabel: "true",
+		}
+		disk1, err := computeService.Disks.Get(p, zones[0], volName).Do()
+		Expect(err).To(BeNil(), "Could not get disk")
+		disk1Op, err := computeService.Disks.SetLabels(p, zones[0], volName, &compute.ZoneSetLabelsRequest{
+			LabelFingerprint: disk1.LabelFingerprint,
+			Labels:           labelsMap,
+		}).Do()
+		Expect(err).To(BeNil(), "Could not set disk labels")
+		_, err = computeService.ZoneOperations.Wait(p, zones[0], disk1Op.Name).Do()
+		Expect(err).To(BeNil(), "Could not set disk labels")
+
+		defer deleteDisk(controllerClient, p, zones[0], volID0, volName)
+
+		// Create multi-zone Disk
+		resp, err := controllerClient.CreateVolumeWithCaps(volName, map[string]string{
+			common.ParameterKeyEnableMultiZoneProvisioning: "true",
+			common.ParameterKeyType:                        "hyperdisk-ml",
+		}, defaultHdmlSizeGb,
+			&csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{common.TopologyKeyZone: zones[1]},
+					},
+				},
+			},
+			[]*csi.VolumeCapability{
+				{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+					},
+				},
+			},
+			nil)
+		Expect(err).To(BeNil(), "Error creating multi-zone volume")
+		topology := resp.GetAccessibleTopology()
+		Expect(len(topology)).To(Equal(2))
+		gotZones := []string{topology[0].Segments[common.TopologyKeyZone], topology[1].Segments[common.TopologyKeyZone]}
+		Expect(gotZones).To(ConsistOf(zones[0], zones[1]))
+
+		volID := fmt.Sprintf("projects/%s/zones/multi-zone/disks/%s", p, volName)
+		defer func() {
+			// Delete Disk
+			err := controllerClient.DeleteVolume(volID)
+			Expect(err).To(BeNil(), "DeleteVolume failed")
+
+			// Validate Disk Deleted
+			_, err = computeService.Disks.Get(p, zones[0], volName).Do()
+			Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found. Err: %v", err)
+			_, err = computeService.Disks.Get(p, zones[1], volName).Do()
+			Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found. Err: %v", err)
+		}()
+
+		disk1, err = computeService.Disks.Get(p, zones[0], volName).Do()
+		Expect(err).To(BeNil(), "Failed to get disk %v/%v", zones[0], volName)
+		disk2, err := computeService.Disks.Get(p, zones[1], volName).Do()
+		Expect(err).To(BeNil(), "Failed to get disk %v/%v", zones[1], volName)
+
+		// Validate disks are RWO
+		Expect(disk1.AccessMode).To(Equal("READ_WRITE_SINGLE"))
+		Expect(disk2.AccessMode).To(Equal("READ_WRITE_SINGLE"))
+	})
+
+	It("Should create ROX 'multi-zone' PV from existing snapshot", func() {
+		checkSkipMultiZoneTests()
+		Expect(testContexts).NotTo(BeEmpty())
+
+		zoneToContext := map[string]*remote.TestContext{}
+		zones := []string{}
+
+		for _, tc := range testContexts {
+			_, z, _ := tc.Instance.GetIdentity()
+			// Zone hasn't been seen before
+			if _, ok := zoneToContext[z]; !ok {
+				zoneToContext[z] = tc
+				zones = append(zones, z)
+			}
+			if len(zoneToContext) == 2 {
+				break
+			}
+		}
+
+		Expect(len(zoneToContext)).To(Equal(2), "Must have instances in 2 zones")
+
+		controllerContext := zoneToContext[zones[0]]
+		controllerClient := controllerContext.Client
+		controllerInstance := controllerContext.Instance
+
+		p, _, _ := controllerInstance.GetIdentity()
+
+		tc0 := zoneToContext[zones[0]]
+		tc1 := zoneToContext[zones[1]]
+
+		snapshotVolName, snapshotVolID := createAndValidateUniqueZonalDisk(controllerClient, p, zones[0], standardDiskType)
+
+		underSpecifiedID := common.GenerateUnderspecifiedVolumeID(snapshotVolName, true /* isZonal */)
+
+		defer func() {
+			// Delete Disk
+			err := controllerClient.DeleteVolume(underSpecifiedID)
+			Expect(err).To(BeNil(), "DeleteVolume failed")
+
+			// Validate Disk Deleted
+			_, err = computeService.Disks.Get(p, zones[0], snapshotVolName).Do()
+			Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found")
+		}()
+
+		// Attach Disk
+		err := testAttachWriteReadDetach(underSpecifiedID, snapshotVolName, tc0.Instance, controllerClient, false /* readOnly */)
+		Expect(err).To(BeNil(), "Failed to go through volume lifecycle")
+
+		// Create Snapshot
+		snapshotName := testNamePrefix + string(uuid.NewUUID())
+		snapshotID, err := controllerClient.CreateSnapshot(snapshotName, snapshotVolID, nil)
+		Expect(err).To(BeNil(), "CreateSnapshot failed with error: %v", err)
+
+		// Validate Snapshot Created
+		snapshot, err := computeService.Snapshots.Get(p, snapshotName).Do()
+		Expect(err).To(BeNil(), "Could not get snapshot from cloud directly")
+		Expect(snapshot.Name).To(Equal(snapshotName))
+
+		err = wait.Poll(10*time.Second, 3*time.Minute, func() (bool, error) {
+			snapshot, err := computeService.Snapshots.Get(p, snapshotName).Do()
+			Expect(err).To(BeNil(), "Could not get snapshot from cloud directly")
+			if snapshot.Status == "READY" {
+				return true, nil
+			}
+			return false, nil
+		})
+		Expect(err).To(BeNil(), "Could not wait for snapshot be ready")
+
+		// Create multi-zone Disk
+		volName := testNamePrefix + string(uuid.NewUUID())
+		_, err = controllerClient.CreateVolumeWithCaps(volName, map[string]string{
+			common.ParameterKeyEnableMultiZoneProvisioning: "true",
+			common.ParameterKeyType:                        "hyperdisk-ml",
+		}, defaultHdmlSizeGb,
+			&csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{common.TopologyKeyZone: zones[0]},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: zones[1]},
+					},
+				},
+			},
+			[]*csi.VolumeCapability{
+				{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY,
+					},
+				},
+			},
+			&csi.VolumeContentSource{
+				Type: &csi.VolumeContentSource_Snapshot{
+					Snapshot: &csi.VolumeContentSource_SnapshotSource{
+						SnapshotId: snapshotID,
+					},
+				},
+			})
+		Expect(err).To(BeNil(), "CreateVolume failed with error: %v", err)
+
+		volID := fmt.Sprintf("projects/%s/zones/multi-zone/disks/%s", p, volName)
+		defer func() {
+			// Delete Disk
+			err := controllerClient.DeleteVolume(volID)
+			Expect(err).To(BeNil(), "DeleteVolume failed")
+
+			// Validate Disk Deleted
+			_, err = computeService.Disks.Get(p, zones[0], volName).Do()
+			Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found. Err: %v", err)
+			_, err = computeService.Disks.Get(p, zones[1], volName).Do()
+			Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found. Err: %v", err)
+		}()
+
+		disk1, err := computeService.Disks.Get(p, zones[0], volName).Do()
+		Expect(err).To(BeNil(), "Failed to get disk %v/%v", zones[0], volName)
+		disk2, err := computeService.Disks.Get(p, zones[1], volName).Do()
+		Expect(err).To(BeNil(), "Failed to get disk %v/%v", zones[1], volName)
+
+		// Validate disks are ROX
+		Expect(disk1.AccessMode).To(Equal("READ_ONLY_MANY"))
+		Expect(disk2.AccessMode).To(Equal("READ_ONLY_MANY"))
+
+		// Attach Disk to node1 and validate contents
+		err = testAttachWriteReadDetach(volID, volName, tc0.Instance, tc0.Client, true /* readonly */)
+		Expect(err).To(BeNil(), "Failed to attach/read/detach on vol1")
+
+		// Attach Disk to node1 and validate contents
+		err = testAttachWriteReadDetach(volID, volName, tc1.Instance, tc1.Client, true /* readonly */)
+		Expect(err).To(BeNil(), "Failed to attach/read/detach on vol2")
+
+		disk1, err = computeService.Disks.Get(p, zones[0], volName).Do()
+		Expect(err).To(BeNil(), "Failed to get disk %v/%v", zones[0], volName)
+		disk2, err = computeService.Disks.Get(p, zones[1], volName).Do()
+		Expect(err).To(BeNil(), "Failed to get disk %v/%v", zones[1], volName)
+
+		// Validate disks have multi-zone labels
+		Expect(disk1.Labels[common.MultiZoneLabel]).To(Equal("true"))
+		Expect(disk2.Labels[common.MultiZoneLabel]).To(Equal("true"))
+
+		// Validate disks are ROX
+		Expect(disk1.AccessMode).To(Equal("READ_ONLY_MANY"))
+		Expect(disk2.AccessMode).To(Equal("READ_ONLY_MANY"))
+	})
+
+	It("Should create ROX 'multi-zone' PV from existing snapshot with no topology", func() {
+		checkSkipMultiZoneTests()
+		Expect(testContexts).NotTo(BeEmpty())
+
+		zoneToContext := map[string]*remote.TestContext{}
+		zones := []string{}
+
+		for _, tc := range testContexts {
+			_, z, _ := tc.Instance.GetIdentity()
+			// Zone hasn't been seen before
+			if _, ok := zoneToContext[z]; !ok {
+				zoneToContext[z] = tc
+				zones = append(zones, z)
+			}
+			if len(zoneToContext) == 2 {
+				break
+			}
+		}
+
+		Expect(len(zoneToContext)).To(Equal(2), "Must have instances in 2 zones")
+
+		controllerContext := zoneToContext[zones[0]]
+		controllerClient := controllerContext.Client
+		controllerInstance := controllerContext.Instance
+
+		p, _, _ := controllerInstance.GetIdentity()
+
+		tc0 := zoneToContext[zones[0]]
+		tc1 := zoneToContext[zones[1]]
+
+		snapshotVolName, snapshotVolID := createAndValidateUniqueZonalDisk(controllerClient, p, zones[0], standardDiskType)
+
+		underSpecifiedID := common.GenerateUnderspecifiedVolumeID(snapshotVolName, true /* isZonal */)
+
+		defer func() {
+			// Delete Disk
+			err := controllerClient.DeleteVolume(underSpecifiedID)
+			Expect(err).To(BeNil(), "DeleteVolume failed")
+
+			// Validate Disk Deleted
+			_, err = computeService.Disks.Get(p, zones[0], snapshotVolName).Do()
+			Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found")
+		}()
+
+		// Attach Disk
+		err := testAttachWriteReadDetach(underSpecifiedID, snapshotVolName, tc0.Instance, controllerClient, false /* readOnly */)
+		Expect(err).To(BeNil(), "Failed to go through volume lifecycle")
+
+		// Create Snapshot
+		snapshotName := testNamePrefix + string(uuid.NewUUID())
+		snapshotID, err := controllerClient.CreateSnapshot(snapshotName, snapshotVolID, nil)
+		Expect(err).To(BeNil(), "CreateSnapshot failed with error: %v", err)
+
+		// Validate Snapshot Created
+		snapshot, err := computeService.Snapshots.Get(p, snapshotName).Do()
+		Expect(err).To(BeNil(), "Could not get snapshot from cloud directly")
+		Expect(snapshot.Name).To(Equal(snapshotName))
+
+		err = wait.Poll(10*time.Second, 3*time.Minute, func() (bool, error) {
+			snapshot, err := computeService.Snapshots.Get(p, snapshotName).Do()
+			Expect(err).To(BeNil(), "Could not get snapshot from cloud directly")
+			if snapshot.Status == "READY" {
+				return true, nil
+			}
+			return false, nil
+		})
+		Expect(err).To(BeNil(), "Could not wait for snapshot be ready")
+
+		// Create multi-zone Disk
+		volName := testNamePrefix + string(uuid.NewUUID())
+		_, err = controllerClient.CreateVolumeWithCaps(volName, map[string]string{
+			common.ParameterKeyEnableMultiZoneProvisioning: "true",
+			common.ParameterKeyType:                        "hyperdisk-ml",
+		}, defaultHdmlSizeGb,
+			&csi.TopologyRequirement{},
+			[]*csi.VolumeCapability{
+				{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY,
+					},
+				},
+			},
+			&csi.VolumeContentSource{
+				Type: &csi.VolumeContentSource_Snapshot{
+					Snapshot: &csi.VolumeContentSource_SnapshotSource{
+						SnapshotId: snapshotID,
+					},
+				},
+			})
+		Expect(err).To(BeNil(), "CreateVolume failed with error: %v", err)
+
+		volID := fmt.Sprintf("projects/%s/zones/multi-zone/disks/%s", p, volName)
+		defer func() {
+			// Delete Disk
+			err := controllerClient.DeleteVolume(volID)
+			Expect(err).To(BeNil(), "DeleteVolume failed")
+
+			// Validate Disk Deleted
+			_, err = computeService.Disks.Get(p, zones[0], volName).Do()
+			Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found. Err: %v", err)
+			_, err = computeService.Disks.Get(p, zones[1], volName).Do()
+			Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found. Err: %v", err)
+		}()
+
+		disk1, err := computeService.Disks.Get(p, zones[0], volName).Do()
+		Expect(err).To(BeNil(), "Failed to get disk %v/%v", zones[0], volName)
+		disk2, err := computeService.Disks.Get(p, zones[1], volName).Do()
+		Expect(err).To(BeNil(), "Failed to get disk %v/%v", zones[1], volName)
+
+		// Validate disks are ROX
+		Expect(disk1.AccessMode).To(Equal("READ_ONLY_MANY"))
+		Expect(disk2.AccessMode).To(Equal("READ_ONLY_MANY"))
+
+		// Attach Disk to node1 and validate contents
+		err = testAttachWriteReadDetach(volID, volName, tc0.Instance, tc0.Client, true /* readonly */)
+		Expect(err).To(BeNil(), "Failed to attach/read/detach on vol1")
+
+		// Attach Disk to node1 and validate contents
+		err = testAttachWriteReadDetach(volID, volName, tc1.Instance, tc1.Client, true /* readonly */)
+		Expect(err).To(BeNil(), "Failed to attach/read/detach on vol2")
+
+		disk1, err = computeService.Disks.Get(p, zones[0], volName).Do()
+		Expect(err).To(BeNil(), "Failed to get disk %v/%v", zones[0], volName)
+		disk2, err = computeService.Disks.Get(p, zones[1], volName).Do()
+		Expect(err).To(BeNil(), "Failed to get disk %v/%v", zones[1], volName)
+
+		// Validate disks have multi-zone labels
+		Expect(disk1.Labels[common.MultiZoneLabel]).To(Equal("true"))
+		Expect(disk2.Labels[common.MultiZoneLabel]).To(Equal("true"))
+
+		// Validate disks are ROX
+		Expect(disk1.AccessMode).To(Equal("READ_ONLY_MANY"))
+		Expect(disk2.AccessMode).To(Equal("READ_ONLY_MANY"))
+	})
+
+	It("Should create ROX 'multi-zone' PV from existing disk image", func() {
+		checkSkipMultiZoneTests()
+		Expect(testContexts).NotTo(BeEmpty())
+
+		zoneToContext := map[string]*remote.TestContext{}
+		zones := []string{}
+
+		for _, tc := range testContexts {
+			_, z, _ := tc.Instance.GetIdentity()
+			// Zone hasn't been seen before
+			if _, ok := zoneToContext[z]; !ok {
+				zoneToContext[z] = tc
+				zones = append(zones, z)
+			}
+			if len(zoneToContext) == 2 {
+				break
+			}
+		}
+
+		Expect(len(zoneToContext)).To(Equal(2), "Must have instances in 2 zones")
+
+		controllerContext := zoneToContext[zones[0]]
+		controllerClient := controllerContext.Client
+		controllerInstance := controllerContext.Instance
+
+		p, _, _ := controllerInstance.GetIdentity()
+
+		tc0 := zoneToContext[zones[0]]
+		tc1 := zoneToContext[zones[1]]
+
+		snapshotVolName, snapshotVolID := createAndValidateUniqueZonalDisk(controllerClient, p, zones[0], standardDiskType)
+
+		underSpecifiedID := common.GenerateUnderspecifiedVolumeID(snapshotVolName, true /* isZonal */)
+
+		defer func() {
+			// Delete Disk
+			err := controllerClient.DeleteVolume(underSpecifiedID)
+			Expect(err).To(BeNil(), "DeleteVolume failed")
+
+			// Validate Disk Deleted
+			_, err = computeService.Disks.Get(p, zones[0], snapshotVolName).Do()
+			Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found")
+		}()
+
+		// Attach Disk
+		err := testAttachWriteReadDetach(underSpecifiedID, snapshotVolName, tc0.Instance, controllerClient, false /* readOnly */)
+		Expect(err).To(BeNil(), "Failed to go through volume lifecycle")
+
+		// Create Disk Image
+		imageName := testNamePrefix + string(uuid.NewUUID())
+		snapshotParams := map[string]string{
+			"snapshot-type": "images",
+		}
+		snapshotID, err := controllerClient.CreateSnapshot(imageName, snapshotVolID, snapshotParams)
+		klog.Infof("Created image snapshot with snapshotID: %s", snapshotID)
+		Expect(err).To(BeNil(), "CreateSnapshot failed with error: %v", err)
+
+		// Validate Disk Image Created
+		image, err := computeService.Images.Get(p, imageName).Do()
+		Expect(err).To(BeNil(), "Could not get disk image from cloud directly")
+		Expect(image.Name).To(Equal(imageName))
+
+		err = wait.Poll(10*time.Second, 3*time.Minute, func() (bool, error) {
+			image, err := computeService.Images.Get(p, imageName).Do()
+			Expect(err).To(BeNil(), "Could not get disk image from cloud directly")
+			if image.Status == "READY" {
+				return true, nil
+			}
+			return false, nil
+		})
+		Expect(err).To(BeNil(), "Could not wait for disk image be ready")
+
+		// Create multi-zone Disk
+		volName := testNamePrefix + string(uuid.NewUUID())
+
+		_, err = controllerClient.CreateVolumeWithCaps(volName, map[string]string{
+			common.ParameterKeyEnableMultiZoneProvisioning: "true",
+			common.ParameterKeyType:                        "hyperdisk-ml",
+		}, defaultHdmlSizeGb,
+			&csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{common.TopologyKeyZone: zones[0]},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: zones[1]},
+					},
+				},
+			},
+			[]*csi.VolumeCapability{
+				{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY,
+					},
+				},
+			},
+			&csi.VolumeContentSource{
+				Type: &csi.VolumeContentSource_Snapshot{
+					Snapshot: &csi.VolumeContentSource_SnapshotSource{
+						SnapshotId: snapshotID,
+					},
+				},
+			})
+		Expect(err).To(BeNil(), "CreateVolume failed with error: %v", err)
+
+		volID := fmt.Sprintf("projects/%s/zones/multi-zone/disks/%s", p, volName)
+		defer func() {
+			// Delete Disk
+			err := controllerClient.DeleteVolume(volID)
+			Expect(err).To(BeNil(), "DeleteVolume failed")
+
+			// Validate Disk Deleted
+			_, err = computeService.Disks.Get(p, zones[0], volName).Do()
+			Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found. Err: %v", err)
+			_, err = computeService.Disks.Get(p, zones[1], volName).Do()
+			Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found. Err: %v", err)
+		}()
+
+		disk1, err := computeService.Disks.Get(p, zones[0], volName).Do()
+		Expect(err).To(BeNil(), "Failed to get disk %v/%v", zones[0], volName)
+		disk2, err := computeService.Disks.Get(p, zones[1], volName).Do()
+		Expect(err).To(BeNil(), "Failed to get disk %v/%v", zones[1], volName)
+
+		// Validate disks have multi-zone labels
+		Expect(disk1.Labels[common.MultiZoneLabel]).To(Equal("true"))
+		Expect(disk2.Labels[common.MultiZoneLabel]).To(Equal("true"))
+
+		// Validate disks are ROX
+		Expect(disk1.AccessMode).To(Equal("READ_ONLY_MANY"))
+		Expect(disk2.AccessMode).To(Equal("READ_ONLY_MANY"))
+
+		// Attach Disk to node1
+		err = testAttachWriteReadDetach(volID, volName, tc0.Instance, tc0.Client, true /* readonly */)
+		Expect(err).To(BeNil(), "Failed to attach/read/detach on vol1")
+
+		// Attach Disk to node1
+		err = testAttachWriteReadDetach(volID, volName, tc1.Instance, tc1.Client, true /* readonly */)
+		Expect(err).To(BeNil(), "Failed to attach/read/detach on vol2")
+	})
+
+	It("Should create RWO 'multi-zone' PV that has empty disks", func() {
+		checkSkipMultiZoneTests()
+		// Create new driver and client
+		Expect(testContexts).NotTo(BeEmpty())
+
+		zoneToContext := map[string]*remote.TestContext{}
+		zones := []string{}
+
+		for _, tc := range testContexts {
+			_, z, _ := tc.Instance.GetIdentity()
+			// Zone hasn't been seen before
+			if _, ok := zoneToContext[z]; !ok {
+				zoneToContext[z] = tc
+				zones = append(zones, z)
+			}
+			if len(zoneToContext) == 2 {
+				break
+			}
+		}
+
+		Expect(len(zoneToContext)).To(Equal(2), "Must have instances in 2 zones")
+
+		controllerContext := zoneToContext[zones[0]]
+		controllerClient := controllerContext.Client
+		controllerInstance := controllerContext.Instance
+
+		p, _, _ := controllerInstance.GetIdentity()
+
+		// Attach disk to instance in the first zone.
+		tc0 := zoneToContext[zones[0]]
+		tc1 := zoneToContext[zones[1]]
+
+		// Create Disk
+		volName := testNamePrefix + string(uuid.NewUUID())
+		_, err := controllerClient.CreateVolumeWithCaps(volName, map[string]string{
+			common.ParameterKeyEnableMultiZoneProvisioning: "true",
+			common.ParameterKeyType:                        "hyperdisk-ml",
+		}, defaultHdmlSizeGb,
+			&csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{common.TopologyKeyZone: zones[0]},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: zones[1]},
+					},
+				},
+			},
+			[]*csi.VolumeCapability{
+				{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+					},
+				},
+			},
+			nil)
+		Expect(err).To(BeNil(), "CreateVolume failed with error: %v", err)
+
+		volID := fmt.Sprintf("projects/%s/zones/multi-zone/disks/%s", p, volName)
+		defer func() {
+			// Delete Disk
+			err := controllerClient.DeleteVolume(volID)
+			Expect(err).To(BeNil(), "DeleteVolume failed")
+
+			// Validate Disk Deleted
+			_, err = computeService.Disks.Get(p, zones[0], volName).Do()
+			Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found. Err: %v", err)
+			_, err = computeService.Disks.Get(p, zones[1], volName).Do()
+			Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found. Err: %v", err)
+		}()
+
+		disk1, err := computeService.Disks.Get(p, zones[0], volName).Do()
+		Expect(err).To(BeNil(), "Failed to get disk %v/%v", zones[0], volName)
+		disk2, err := computeService.Disks.Get(p, zones[1], volName).Do()
+		Expect(err).To(BeNil(), "Failed to get disk %v/%v", zones[1], volName)
+
+		// Validate disks have multi-zone labels
+		Expect(disk1.Labels[common.MultiZoneLabel]).To(Equal("true"))
+		Expect(disk2.Labels[common.MultiZoneLabel]).To(Equal("true"))
+
+		// Validate disks are RWO
+		Expect(disk1.AccessMode).To(Equal("READ_WRITE_SINGLE"))
+		Expect(disk2.AccessMode).To(Equal("READ_WRITE_SINGLE"))
+
+		// Validate underlying disks can be used
+		volID0 := fmt.Sprintf("projects/%s/zones/%s/disks/%s", p, zones[0], volName)
+		volID1 := fmt.Sprintf("projects/%s/zones/%s/disks/%s", p, zones[1], volName)
+
+		err = testAttachWriteReadDetach(volID0, volName, tc0.Instance, tc0.Client, false /* readonly */)
+		Expect(err).To(BeNil(), "Failed to attach/write/read/detach on vol1")
+
+		err = testAttachWriteReadDetach(volID1, volName, tc1.Instance, tc1.Client, false /* readonly */)
+		Expect(err).To(BeNil(), "Failed to attach/write/read/detach on vol2")
+
+		// Validate disks can be used in multi-zone mode on both nodes
+		volIDMultiZone := fmt.Sprintf("projects/%s/zones/multi-zone/disks/%s", p, volName)
+		err = testAttachWriteReadDetach(volIDMultiZone, volName, tc0.Instance, tc0.Client, true /* readonly */)
+		Expect(err).To(BeNil(), "Failed to attach/read/detach on vol1")
+
+		err = testAttachWriteReadDetach(volIDMultiZone, volName, tc1.Instance, tc1.Client, true /* readonly */)
+		Expect(err).To(BeNil(), "Failed to attach/read/detach on vol2")
+
+		// Validate disks are ROX now
+		disk1, err = computeService.Disks.Get(p, zones[0], volName).Do()
+		Expect(err).To(BeNil(), "Failed to get disk %v/%v", zones[0], volName)
+		disk2, err = computeService.Disks.Get(p, zones[1], volName).Do()
+		Expect(err).To(BeNil(), "Failed to get disk %v/%v", zones[1], volName)
+
+		Expect(disk1.AccessMode).To(Equal("READ_ONLY_MANY"))
+		Expect(disk2.AccessMode).To(Equal("READ_ONLY_MANY"))
 
 	})
 
diff --git a/test/e2e/tests/setup_e2e_test.go b/test/e2e/tests/setup_e2e_test.go
index 0fd6f1a7..7fd12a52 100644
--- a/test/e2e/tests/setup_e2e_test.go
+++ b/test/e2e/tests/setup_e2e_test.go
@@ -126,6 +126,7 @@ func notEmpty(v string) bool {
 func getDriverConfig() testutils.DriverConfig {
 	return testutils.DriverConfig{
 		ExtraFlags: slices.Filter(nil, strings.Split(*extraDriverFlags, ","), notEmpty),
+		Zones:      strings.Split(*zones, ","),
 	}
 }
 
diff --git a/test/e2e/tests/single_zone_e2e_test.go b/test/e2e/tests/single_zone_e2e_test.go
index 980bb0f7..5bb98944 100644
--- a/test/e2e/tests/single_zone_e2e_test.go
+++ b/test/e2e/tests/single_zone_e2e_test.go
@@ -47,6 +47,7 @@ const (
 	defaultSizeGb                     int64 = 5
 	defaultExtremeSizeGb              int64 = 500
 	defaultHdTSizeGb                  int64 = 2048
+	defaultHdmlSizeGb                 int64 = 200
 	defaultRepdSizeGb                 int64 = 200
 	defaultMwSizeGb                   int64 = 200
 	defaultVolumeLimit                int64 = 127
@@ -56,6 +57,7 @@ const (
 	ssdDiskType                             = "pd-ssd"
 	extremeDiskType                         = "pd-extreme"
 	hdtDiskType                             = "hyperdisk-throughput"
+	hdmlDiskType                            = "hyperdisk-ml"
 	provisionedIOPSOnCreate                 = "12345"
 	provisionedIOPSOnCreateInt              = int64(12345)
 	provisionedIOPSOnCreateDefaultInt       = int64(100000)
@@ -1579,6 +1581,8 @@ func createAndValidateZonalDisk(client *remote.CsiClient, project, zone string,
 		diskSize = defaultExtremeSizeGb
 	case hdtDiskType:
 		diskSize = defaultHdTSizeGb
+	case hdmlDiskType:
+		diskSize = defaultHdmlSizeGb
 	}
 	volume, err := client.CreateVolume(volName, disk.params, diskSize,
 		&csi.TopologyRequirement{
@@ -1759,6 +1763,14 @@ var typeToDisk = map[string]*disk{
 			Expect(disk.Type).To(ContainSubstring(ssdDiskType))
 		},
 	},
+	"hyperdisk-ml": {
+		params: map[string]string{
+			common.ParameterKeyType: "hyperdisk-ml",
+		},
+		validate: func(disk *compute.Disk) {
+			Expect(disk.Type).To(ContainSubstring("hyperdisk-ml"))
+		},
+	},
 }
 
 func merge(a, b map[string]string) map[string]string {
diff --git a/test/e2e/utils/utils.go b/test/e2e/utils/utils.go
index 3d4a2df8..995b114b 100644
--- a/test/e2e/utils/utils.go
+++ b/test/e2e/utils/utils.go
@@ -46,6 +46,7 @@ var (
 type DriverConfig struct {
 	ComputeEndpoint string
 	ExtraFlags      []string
+	Zones           []string
 }
 
 func GCEClientAndDriverSetup(instance *remote.InstanceInfo, driverConfig DriverConfig) (*remote.TestContext, error) {
@@ -62,9 +63,10 @@ func GCEClientAndDriverSetup(instance *remote.InstanceInfo, driverConfig DriverC
 		fmt.Sprintf("--extra-labels=%s=%s", DiskLabelKey, DiskLabelValue),
 		"--max-concurrent-format-and-mount=20", // otherwise the serialization times out the e2e test.
 		"--multi-zone-volume-handle-enable",
-		"--multi-zone-volume-handle-disk-types=pd-standard",
+		"--multi-zone-volume-handle-disk-types=pd-standard,hyperdisk-ml",
 		"--use-instance-api-to-poll-attachment-disk-types=pd-ssd",
 		"--use-instance-api-to-list-volumes-published-nodes",
+		fmt.Sprintf("--fallback-requisite-zones=%s", strings.Join(driverConfig.Zones, ",")),
 	}
 	extra_flags = append(extra_flags, fmt.Sprintf("--compute-endpoint=%s", driverConfig.ComputeEndpoint))
 	extra_flags = append(extra_flags, driverConfig.ExtraFlags...)