Repository: kubernetes-sigs/gcp-compute-persistent-disk-csi-driver
Pull request #1696 (2024-05-03 21:24:02)

Use instances.list to determine PublishedNodeIds in ListVolume API
  **What type of PR is this?**
  /kind bug
  
  **What this PR does / why we need it**:
  
  The `disks.list` API is currently used in ListVolumes RPC. This looks up all disks within a project/region. The response includes a `items[].users[]` field, which contains the list of all VMs that are attached to a given disk. Thehe `items[].users[]` array has an implicit limit of 50 VMs. If a disk is attached to more than 50 instances, only 50 instances will be returned in `items[].users[]`. This causes problems with the [`external-attacher`](https://github.com/kubernetes-csi/external-attacher/blob/5f016e3d49b561d4e72375468684c6e19e56044a/pkg/controller/csi_handler.go#L134) reconciliation logic, since these disks appear to not be attached. When this occurs, the `external-attacher` will re-issue unnecessary ControllerPublishVolume calls for these disk/VM pairs. This causes additional RPC calls to the compute API (2x per ControllerPublish for `instances.get` and `disks.get`), and increases load on the PDCSI controller. The user facing effect is that GCP quota can be consumed unnecessarily.
  
The `instances.list` API has a `items[].disks[]` field which indicates all attached disks for an instance.  By moving to `instances.list` API to determine which VMs a disk is attached to, this gets around the 50 instance limit. Each VM can attach up to 128 disks (127 excluding the boot disk), so the `instances.list` API does not limit responses.
  
By making this new API call, the PDCSI controller will receive more data from the GCE API (from `instances.list` and `instances.get`). This has the potential to increase the latency of the ListVolume RPC call. This PR attempts to offset cost of this additional (paginated) API call by introducing two enhancements:
   *  This PR adds a new command line flag to filter disks that are returned in `instances.list` API call. This allows a driver to limit disks with particular features. For example, in Kubernetes, the driver should only be concerned about determining if disks are attached to VMs associated with the Kubernetes cluster the driver is running in. To limit total responses from `instances.list`, an administrator can pre-tag node VMs with a label (eg: the Kubernetes cluster name) and run the driver with the `--instances-list-filters` flag, targeting this label.
   * This PR limits the response fields returned from the GCE API for `disks.list` and `instances.list` API calls. Only a subset of the available response fields are actually used by the PDCSI controller client. This should reduce the payload response size, reducing the memory burst footprint of ListVolumes API calls (since partial results are cached in memory), and response latency of each paginated call, as less data needs to be transfered from the GCE server endpoint.
  
  **Does this PR introduce a user-facing change?**:
  ```release-note
The --instances-list-filters command line flag allows the driver to filter on response properties when listing VMs from the GCE API in ListVolumes RPC.
The --use-instance-api-to-list-volumes-published-nodes command line flag allows the driver to use the instances.list API instead of disks.list API when determining PublishedNodeIs in ListVolumes RPC.
  ```

**Testing**:

Validated existing behavior with:

```
./test/run-e2e-local.sh --enable-confidential-compute --use-instance-api-to-list-volumes-published-nodes=false
```

Validated new behavior:

```
./test/run-e2e-local.sh --enable-confidential-compute --use-instance-api-to-list-volumes-published-nodes
```


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 | d87da40c3ecb08c90a7ca9702c557a5da7345d8e | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_gcp-compute-persistent-disk-csi-driver/1696/pull-gcp-compute-persistent-disk-csi-driver-verify/1786507039808163840) | 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=1696). [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: [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/1696#" 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 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 | d87da40c3ecb08c90a7ca9702c557a5da7345d8e | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_gcp-compute-persistent-disk-csi-driver/1696/pull-gcp-compute-persistent-disk-csi-driver-verify/1786507039808163840) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-verify`
pull-gcp-compute-persistent-disk-csi-driver-unit | d87da40c3ecb08c90a7ca9702c557a5da7345d8e | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_gcp-compute-persistent-disk-csi-driver/1696/pull-gcp-compute-persistent-disk-csi-driver-unit/1786507039753637888) | 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=1696). [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-unit | d87da40c3ecb08c90a7ca9702c557a5da7345d8e | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_gcp-compute-persistent-disk-csi-driver/1696/pull-gcp-compute-persistent-disk-csi-driver-unit/1786507039753637888) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-unit`
pull-gcp-compute-persistent-disk-csi-driver-verify | 134cbc76cd735cb97d29f94d91a7a7b1adab4e38 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_gcp-compute-persistent-disk-csi-driver/1696/pull-gcp-compute-persistent-disk-csi-driver-verify/1786514308121759744) | 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=1696). [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 | 73e41199c6d5e70f24e61ff8871269d285b16858 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_gcp-compute-persistent-disk-csi-driver/1696/pull-gcp-compute-persistent-disk-csi-driver-e2e/1786514681016356864) | 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=1696). [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 | 73e41199c6d5e70f24e61ff8871269d285b16858 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_gcp-compute-persistent-disk-csi-driver/1696/pull-gcp-compute-persistent-disk-csi-driver-e2e/1786514681016356864) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-e2e`
pull-gcp-compute-persistent-disk-csi-driver-unit | c226daacdc642398168510a9469dcd4517cc5a2a | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_gcp-compute-persistent-disk-csi-driver/1696/pull-gcp-compute-persistent-disk-csi-driver-unit/1787521706462547968) | 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=1696). [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/cmd/gce-pd-csi-driver/main.go b/cmd/gce-pd-csi-driver/main.go
index 59dbfd1c..83666475 100644
--- a/cmd/gce-pd-csi-driver/main.go
+++ b/cmd/gce-pd-csi-driver/main.go
@@ -79,7 +79,9 @@ var (
 	computeEndpoint           *url.URL
 	allowedComputeEnvironment = []gce.Environment{gce.EnvironmentStaging, gce.EnvironmentProduction}
 
-	useInstanceAPIOnWaitForAttachDiskTypesFlag = flag.String("use-instance-api-to-poll-attachment-disk-types", "", "Comma separated list of disk types that should use instances.get API when polling for disk attach during ControllerPublish")
+	useInstanceAPIOnWaitForAttachDiskTypesFlag     = flag.String("use-instance-api-to-poll-attachment-disk-types", "", "Comma separated list of disk types that should use instances.get API when polling for disk attach during ControllerPublish")
+	useInstanceAPIForListVolumesPublishedNodesFlag = flag.Bool("use-instance-api-to-list-volumes-published-nodes", false, "Enables using the instances.list API to determine published_node_ids in ListVolumes. When false (default), the disks.list API is used")
+	instancesListFiltersFlag                       = flag.String("instances-list-filters", "", "Comma separated list of filters to use when calling the instances.list API. By default instances.list fetches all instances in a region")
 
 	extraTagsStr = flag.String("extra-tags", "", "Extra tags to attach to each Compute Disk, Image, Snapshot created. It is a comma separated list of parent id, key and value like '<parent_id1>/<tag_key1>/<tag_value1>,...,<parent_idN>/<tag_keyN>/<tag_valueN>'. parent_id is the Organization or the Project ID or Project name where the tag key and the tag value resources exist. A maximum of 50 tags bindings is allowed for a resource. See https://cloud.google.com/resource-manager/docs/tags/tags-overview, https://cloud.google.com/resource-manager/docs/tags/tags-creating-and-managing for details")
 
@@ -188,16 +190,25 @@ func handle() {
 		UseInstancesAPIForDiskTypes: useInstanceAPIOnWaitForAttachDiskTypes,
 	}
 
+	// Initialize listVolumes config
+	instancesListFilters := strings.Split(*instancesListFiltersFlag, ",")
+	listInstancesConfig := gce.ListInstancesConfig{
+		Filters: instancesListFilters,
+	}
+	listVolumesConfig := driver.ListVolumesConfig{
+		UseInstancesAPIForPublishedNodes: *useInstanceAPIForListVolumesPublishedNodesFlag,
+	}
+
 	// Initialize requirements for the controller service
 	var controllerServer *driver.GCEControllerServer
 	if *runControllerService {
-		cloudProvider, err := gce.CreateCloudProvider(ctx, version, *cloudConfigFilePath, computeEndpoint, computeEnvironment, waitForAttachConfig)
+		cloudProvider, err := gce.CreateCloudProvider(ctx, version, *cloudConfigFilePath, computeEndpoint, computeEnvironment, waitForAttachConfig, listInstancesConfig)
 		if err != nil {
 			klog.Fatalf("Failed to get cloud provider: %v", err.Error())
 		}
 		initialBackoffDuration := time.Duration(*errorBackoffInitialDurationMs) * time.Millisecond
 		maxBackoffDuration := time.Duration(*errorBackoffMaxDurationMs) * time.Millisecond
-		controllerServer = driver.NewControllerServer(gceDriver, cloudProvider, initialBackoffDuration, maxBackoffDuration, fallbackRequisiteZones, *enableStoragePoolsFlag, multiZoneVolumeHandleConfig)
+		controllerServer = driver.NewControllerServer(gceDriver, cloudProvider, initialBackoffDuration, maxBackoffDuration, fallbackRequisiteZones, *enableStoragePoolsFlag, multiZoneVolumeHandleConfig, listVolumesConfig)
 	} else if *cloudConfigFilePath != "" {
 		klog.Warningf("controller service is disabled but cloud config given - it has no effect")
 	}
diff --git a/pkg/deviceutils/device-utils.go b/pkg/deviceutils/device-utils.go
index c9ce8988..134ae0f2 100644
--- a/pkg/deviceutils/device-utils.go
+++ b/pkg/deviceutils/device-utils.go
@@ -86,6 +86,10 @@ type DeviceUtils interface {
 	// exists on the machine, or an empty string if none exists
 	VerifyDevicePath(devicePaths []string, deviceName string) (string, error)
 
+	// DisableDevice performs necessary disabling prior to a device being
+	// detached from a node. The path is that from GetDiskByIdPaths.
+	DisableDevice(devicePath string) error
+
 	// Resize returns whether or not a device needs resizing.
 	Resize(resizer resizefs.Resizefs, devicePath string, deviceMountPath string) (bool, error)
 
diff --git a/pkg/deviceutils/device-utils_linux.go b/pkg/deviceutils/device-utils_linux.go
index 5878a28f..f480359a 100644
--- a/pkg/deviceutils/device-utils_linux.go
+++ b/pkg/deviceutils/device-utils_linux.go
@@ -24,6 +24,11 @@ import (
 	"k8s.io/mount-utils"
 )
 
+func (_ *deviceUtils) DisableDevice(devicePath string) error {
+	deviceName := filepath.Base(devicePath)
+	return os.WriteFile(fmt.Sprintf("/sys/block/%s/device/state", deviceName), []byte("offline"), 0644)
+}
+
 func (_ *deviceUtils) IsDeviceFilesystemInUse(mounter *mount.SafeFormatAndMount, devicePath, devFsPath string) (bool, error) {
 	fstype, err := mounter.GetDiskFormat(devicePath)
 	if err != nil {
diff --git a/pkg/gce-cloud-provider/compute/fake-gce.go b/pkg/gce-cloud-provider/compute/fake-gce.go
index f2f33e76..3b7416b1 100644
--- a/pkg/gce-cloud-provider/compute/fake-gce.go
+++ b/pkg/gce-cloud-provider/compute/fake-gce.go
@@ -127,7 +127,7 @@ func (cloud *FakeCloudProvider) ListZones(ctx context.Context, region string) ([
 	return []string{cloud.zone, "country-region-fakesecondzone"}, nil
 }
 
-func (cloud *FakeCloudProvider) ListDisks(ctx context.Context) ([]*computev1.Disk, string, error) {
+func (cloud *FakeCloudProvider) ListDisks(ctx context.Context, fields []googleapi.Field) ([]*computev1.Disk, string, error) {
 	d := []*computev1.Disk{}
 	for _, cd := range cloud.disks {
 		d = append(d, cd.disk)
@@ -135,6 +135,14 @@ func (cloud *FakeCloudProvider) ListDisks(ctx context.Context) ([]*computev1.Dis
 	return d, "", nil
 }
 
+func (cloud *FakeCloudProvider) ListInstances(ctx context.Context, fields []googleapi.Field) ([]*computev1.Instance, string, error) {
+	instances := []*computev1.Instance{}
+	for _, instance := range cloud.instances {
+		instances = append(instances, instance)
+	}
+	return instances, "", nil
+}
+
 func (cloud *FakeCloudProvider) ListSnapshots(ctx context.Context, filter string) ([]*computev1.Snapshot, string, error) {
 	var sourceDisk string
 	snapshots := []*computev1.Snapshot{}
diff --git a/pkg/gce-cloud-provider/compute/gce-compute.go b/pkg/gce-cloud-provider/compute/gce-compute.go
index 8948ff6c..eaa0c995 100644
--- a/pkg/gce-cloud-provider/compute/gce-compute.go
+++ b/pkg/gce-cloud-provider/compute/gce-compute.go
@@ -33,6 +33,7 @@ import (
 	"golang.org/x/oauth2"
 	computebeta "google.golang.org/api/compute/v0.beta"
 	computev1 "google.golang.org/api/compute/v1"
+	"google.golang.org/api/googleapi"
 	"google.golang.org/api/iterator"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -106,7 +107,8 @@ type GCECompute interface {
 	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) ([]*computev1.Disk, string, error)
+	ListDisks(ctx context.Context, fields []googleapi.Field) ([]*computev1.Disk, string, error)
+	ListInstances(ctx context.Context, fields []googleapi.Field) ([]*computev1.Instance, string, error)
 	// Regional Disk Methods
 	GetReplicaZoneURI(project string, zone string) string
 	// Instance Methods
@@ -135,7 +137,7 @@ 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) ([]*computev1.Disk, string, error) {
+func (cloud *CloudProvider) ListDisks(ctx context.Context, fields []googleapi.Field) ([]*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)
@@ -148,6 +150,7 @@ func (cloud *CloudProvider) ListDisks(ctx context.Context) ([]*computev1.Disk, s
 
 	// listing out regional disks in the region
 	rlCall := cloud.service.RegionDisks.List(cloud.project, region)
+	rlCall.Fields(fields...)
 	nextPageToken := "pageToken"
 	for nextPageToken != "" {
 		rDiskList, err := rlCall.Do()
@@ -162,6 +165,7 @@ func (cloud *CloudProvider) ListDisks(ctx context.Context) ([]*computev1.Disk, s
 	// listing out zonal disks in all zones of the region
 	for _, zone := range zones {
 		lCall := cloud.service.Disks.List(cloud.project, zone)
+		lCall.Fields(fields...)
 		nextPageToken := "pageToken"
 		for nextPageToken != "" {
 			diskList, err := lCall.Do()
@@ -176,6 +180,41 @@ func (cloud *CloudProvider) ListDisks(ctx context.Context) ([]*computev1.Disk, s
 	return items, "", nil
 }
 
+// ListInstances lists instances based on maxEntries and pageToken for the project and region
+// that the driver is running in. Filters from cloud.listInstancesConfig.Filters are applied
+// to the request.
+func (cloud *CloudProvider) ListInstances(ctx context.Context, fields []googleapi.Field) ([]*computev1.Instance, string, error) {
+	region, err := common.GetRegionFromZones([]string{cloud.zone})
+	if err != nil {
+		return nil, "", fmt.Errorf("failed to get region from zones: %w", err)
+	}
+	zones, err := cloud.ListZones(ctx, region)
+	if err != nil {
+		return nil, "", err
+	}
+	items := []*computev1.Instance{}
+
+	for _, zone := range zones {
+		lCall := cloud.service.Instances.List(cloud.project, zone)
+		for _, filter := range cloud.listInstancesConfig.Filters {
+			lCall = lCall.Filter(filter)
+		}
+		lCall = lCall.Fields(fields...)
+		nextPageToken := "pageToken"
+		for nextPageToken != "" {
+			instancesList, err := lCall.Do()
+			if err != nil {
+				return nil, "", err
+			}
+			items = append(items, instancesList.Items...)
+			nextPageToken = instancesList.NextPageToken
+			lCall.PageToken(nextPageToken)
+		}
+	}
+
+	return items, "", nil
+}
+
 // RepairUnderspecifiedVolumeKey will query the cloud provider and check each zone for the disk specified
 // by the volume key and return a volume key with a correct zone
 func (cloud *CloudProvider) RepairUnderspecifiedVolumeKey(ctx context.Context, project string, volumeKey *meta.Key) (string, *meta.Key, error) {
diff --git a/pkg/gce-cloud-provider/compute/gce.go b/pkg/gce-cloud-provider/compute/gce.go
index 3f553a93..c2d2df8c 100644
--- a/pkg/gce-cloud-provider/compute/gce.go
+++ b/pkg/gce-cloud-provider/compute/gce.go
@@ -102,6 +102,8 @@ type CloudProvider struct {
 	waitForAttachConfig WaitForAttachConfig
 
 	tagsRateLimiter *rate.Limiter
+
+	listInstancesConfig ListInstancesConfig
 }
 
 var _ GCECompute = &CloudProvider{}
@@ -110,6 +112,10 @@ type ConfigFile struct {
 	Global ConfigGlobal `gcfg:"global"`
 }
 
+type ListInstancesConfig struct {
+	Filters []string
+}
+
 type WaitForAttachConfig struct {
 	// A set of disk types that should use the compute instances.get API instead of the
 	// disks.get API. For certain disk types, using the instances.get API is preferred
@@ -128,7 +134,7 @@ type ConfigGlobal struct {
 	Zone      string `gcfg:"zone"`
 }
 
-func CreateCloudProvider(ctx context.Context, vendorVersion string, configPath string, computeEndpoint *url.URL, computeEnvironment Environment, waitForAttachConfig WaitForAttachConfig) (*CloudProvider, error) {
+func CreateCloudProvider(ctx context.Context, vendorVersion string, configPath string, computeEndpoint *url.URL, computeEnvironment Environment, waitForAttachConfig WaitForAttachConfig, listInstancesConfig ListInstancesConfig) (*CloudProvider, error) {
 	configFile, err := readConfig(configPath)
 	if err != nil {
 		return nil, err
@@ -168,6 +174,7 @@ func CreateCloudProvider(ctx context.Context, vendorVersion string, configPath s
 		zone:                zone,
 		zonesCache:          make(map[string]([]string)),
 		waitForAttachConfig: waitForAttachConfig,
+		listInstancesConfig: listInstancesConfig,
 		// GCP has a rate limit of 600 requests per minute, restricting
 		// here to 8 requests per second.
 		tagsRateLimiter: common.NewLimiter(gcpTagsRequestRateLimit, gcpTagsRequestTokenBucketSize, true),
diff --git a/pkg/gce-pd-csi-driver/controller.go b/pkg/gce-pd-csi-driver/controller.go
index 3f6931d3..1bcfdf25 100644
--- a/pkg/gce-pd-csi-driver/controller.go
+++ b/pkg/gce-pd-csi-driver/controller.go
@@ -28,6 +28,7 @@ import (
 	"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud/meta"
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	compute "google.golang.org/api/compute/v1"
+	"google.golang.org/api/googleapi"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/timestamppb"
@@ -105,6 +106,8 @@ type GCEControllerServer struct {
 	enableStoragePools bool
 
 	multiZoneVolumeHandleConfig MultiZoneVolumeHandleConfig
+
+	listVolumesConfig ListVolumesConfig
 }
 
 type MultiZoneVolumeHandleConfig struct {
@@ -124,6 +127,21 @@ type MultiZoneVolumeHandleConfig struct {
 	Enable bool
 }
 
+type ListVolumesConfig struct {
+	UseInstancesAPIForPublishedNodes bool
+}
+
+func (c ListVolumesConfig) listDisksFields() []googleapi.Field {
+	if c.UseInstancesAPIForPublishedNodes {
+		// If we are using the instances.list API in ListVolumes,
+		// don't include the users field in the response, as an optimization.
+		// We rely on instances.list items.disks for attachment pairings.
+		return listDisksFieldsWithoutUsers
+	}
+
+	return listDisksFieldsWithUsers
+}
+
 type csiErrorBackoffId string
 
 type csiErrorBackoff struct {
@@ -175,10 +193,28 @@ const (
 	resourceApiScheme  = "https"
 	resourceApiService = "compute"
 	resourceProject    = "projects"
+
+	listDisksUsersField = googleapi.Field("items/users")
 )
 
 var (
 	validResourceApiVersions = map[string]bool{"v1": true, "alpha": true, "beta": true, "staging_v1": true, "staging_beta": true, "staging_alpha": true}
+
+	// By default GCE returns a lot of data for each instance. Request only a subset of the fields.
+	listInstancesFields = []googleapi.Field{
+		"items/disks/deviceName",
+		"items/disks/source",
+		"items/selfLink",
+		"nextPageToken",
+	}
+
+	// By default GCE returns a lot of data for each disk. Request only a subset of the fields.
+	listDisksFieldsWithoutUsers = []googleapi.Field{
+		"items/labels",
+		"items/selfLink",
+		"nextPageToken",
+	}
+	listDisksFieldsWithUsers = append(listDisksFieldsWithoutUsers, "items/users")
 )
 
 func isDiskReady(disk *gce.CloudDisk) (bool, error) {
@@ -935,6 +971,22 @@ func generateFailedValidationMessage(format string, a ...interface{}) *csi.Valid
 	}
 }
 
+func (gceCS *GCEControllerServer) listVolumeEntries(ctx context.Context) ([]*csi.ListVolumesResponse_Entry, error) {
+	diskList, _, err := gceCS.CloudProvider.ListDisks(ctx, gceCS.listVolumesConfig.listDisksFields())
+	if err != nil {
+		return nil, err
+	}
+
+	var instanceList []*compute.Instance = nil
+	if gceCS.listVolumesConfig.UseInstancesAPIForPublishedNodes {
+		instanceList, _, err = gceCS.CloudProvider.ListInstances(ctx, listInstancesFields)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return gceCS.disksAndInstancesToVolumeEntries(diskList, instanceList), nil
+}
+
 func (gceCS *GCEControllerServer) ListVolumes(ctx context.Context, req *csi.ListVolumesRequest) (*csi.ListVolumesResponse, error) {
 	// https://cloud.google.com/compute/docs/reference/beta/disks/list
 	if req.MaxEntries < 0 {
@@ -944,19 +996,15 @@ func (gceCS *GCEControllerServer) ListVolumes(ctx context.Context, req *csi.List
 
 	offsetLow := 0
 	var ok bool
-	var volumeEntries []*csi.ListVolumesResponse_Entry
 	if req.StartingToken == "" {
-		diskList, _, err := gceCS.CloudProvider.ListDisks(ctx)
+		volumeEntries, err := gceCS.listVolumeEntries(ctx)
 		if err != nil {
 			if gce.IsGCEInvalidError(err) {
 				return nil, status.Errorf(codes.Aborted, "ListVolumes error with invalid request: %v", err.Error())
 			}
-			return nil, common.LoggedError("Failed to list disk: ", err)
+			return nil, common.LoggedError("Failed to list volumes: ", err)
 		}
-		volumeEntries = gceCS.disksToVolumeEntries(diskList)
-	}
 
-	if req.StartingToken == "" {
 		gceCS.volumeEntries = volumeEntries
 		gceCS.volumeEntriesSeen = map[string]int{}
 	} else {
@@ -1008,47 +1056,62 @@ func isMultiZoneDisk(diskRsrc string, diskLabels map[string]string) (string, boo
 	return multiZoneVolumeId, true
 }
 
-// disksToVolumeEntries converts a list of disks to a list of CSI ListVolumeResponse entries
+// disksAndInstancesToVolumeEntries converts a list of disks and instances to a list
+// of CSI ListVolumeResponse entries.
 // It appends "multi-zone" volumeHandles at the end. These are volumeHandles which
 // map to multiple volumeHandles in different zones
-func (gceCS *GCEControllerServer) disksToVolumeEntries(disks []*compute.Disk) []*csi.ListVolumesResponse_Entry {
-	multiZoneNodesByVolumeId := map[string][]string{}
-	entries := []*csi.ListVolumesResponse_Entry{}
+func (gceCS *GCEControllerServer) disksAndInstancesToVolumeEntries(disks []*compute.Disk, instances []*compute.Instance) []*csi.ListVolumesResponse_Entry {
+	nodesByVolumeId := map[string][]string{}
+	multiZoneVolumeIdsByVolumeId := map[string]string{}
 	for _, d := range disks {
-		diskRsrc, err := getResourceId(d.SelfLink)
+		volumeId, err := getResourceId(d.SelfLink)
 		if err != nil {
 			klog.Warningf("Bad ListVolumes disk resource %s, skipped: %v (%+v)", d.SelfLink, err, d)
 			continue
 		}
-		users := []string{}
+
+		instanceIds := make([]string, len(d.Users))
 		for _, u := range d.Users {
-			rsrc, err := getResourceId(u)
+			instanceId, err := getResourceId(u)
 			if err != nil {
 				klog.Warningf("Bad ListVolumes user %s, skipped: %v", u, err)
 			} else {
-				users = append(users, rsrc)
+				instanceIds = append(instanceIds, instanceId)
 			}
 		}
 
+		nodesByVolumeId[volumeId] = instanceIds
+
 		if gceCS.multiZoneVolumeHandleConfig.Enable {
-			if multiZoneVolumeId, isMultiZone := isMultiZoneDisk(diskRsrc, d.Labels); isMultiZone {
-				_, ok := multiZoneNodesByVolumeId[multiZoneVolumeId]
-				if !ok {
-					multiZoneNodesByVolumeId[multiZoneVolumeId] = []string{}
-				}
-				multiZoneNodesByVolumeId[multiZoneVolumeId] = append(multiZoneNodesByVolumeId[multiZoneVolumeId], users...)
+			if multiZoneVolumeId, isMultiZone := isMultiZoneDisk(volumeId, d.Labels); isMultiZone {
+				multiZoneVolumeIdsByVolumeId[volumeId] = multiZoneVolumeId
+				nodesByVolumeId[multiZoneVolumeId] = append(nodesByVolumeId[multiZoneVolumeId], instanceIds...)
 			}
 		}
-		entries = append(entries, &csi.ListVolumesResponse_Entry{
-			Volume: &csi.Volume{
-				VolumeId: diskRsrc,
-			},
-			Status: &csi.ListVolumesResponse_VolumeStatus{
-				PublishedNodeIds: users,
-			},
-		})
 	}
-	for volumeId, nodeIds := range multiZoneNodesByVolumeId {
+
+	entries := []*csi.ListVolumesResponse_Entry{}
+	for _, instance := range instances {
+		instanceId, err := getResourceId(instance.SelfLink)
+		if err != nil {
+			klog.Warningf("Bad ListVolumes instance resource %s, skipped: %v (%+v)", instance.SelfLink, err, instance)
+			continue
+		}
+		for _, disk := range instance.Disks {
+			volumeId, err := getResourceId(disk.Source)
+			if err != nil {
+				klog.Warningf("Bad ListVolumes instance disk source %s, skipped: %v (%+v)", disk.Source, err, instance)
+				continue
+			}
+
+			nodesByVolumeId[volumeId] = append(nodesByVolumeId[volumeId], instanceId)
+			if multiZoneVolumeId, isMultiZone := multiZoneVolumeIdsByVolumeId[volumeId]; isMultiZone {
+				nodesByVolumeId[multiZoneVolumeId] = append(nodesByVolumeId[multiZoneVolumeId], instanceId)
+			}
+		}
+	}
+
+	for volumeId, nodeIds := range nodesByVolumeId {
 		entries = append(entries, &csi.ListVolumesResponse_Entry{
 			Volume: &csi.Volume{
 				VolumeId: volumeId,
diff --git a/pkg/gce-pd-csi-driver/gce-pd-driver.go b/pkg/gce-pd-csi-driver/gce-pd-driver.go
index 96e4a06d..ba1a0c0b 100644
--- a/pkg/gce-pd-csi-driver/gce-pd-driver.go
+++ b/pkg/gce-pd-csi-driver/gce-pd-driver.go
@@ -154,7 +154,7 @@ func NewNodeServer(gceDriver *GCEDriver, mounter *mount.SafeFormatAndMount, devi
 	}
 }
 
-func NewControllerServer(gceDriver *GCEDriver, cloudProvider gce.GCECompute, errorBackoffInitialDuration, errorBackoffMaxDuration time.Duration, fallbackRequisiteZones []string, enableStoragePools bool, multiZoneVolumeHandleConfig MultiZoneVolumeHandleConfig) *GCEControllerServer {
+func NewControllerServer(gceDriver *GCEDriver, cloudProvider gce.GCECompute, errorBackoffInitialDuration, errorBackoffMaxDuration time.Duration, fallbackRequisiteZones []string, enableStoragePools bool, multiZoneVolumeHandleConfig MultiZoneVolumeHandleConfig, listVolumesConfig ListVolumesConfig) *GCEControllerServer {
 	return &GCEControllerServer{
 		Driver:                      gceDriver,
 		CloudProvider:               cloudProvider,
@@ -164,6 +164,7 @@ func NewControllerServer(gceDriver *GCEDriver, cloudProvider gce.GCECompute, err
 		fallbackRequisiteZones:      fallbackRequisiteZones,
 		enableStoragePools:          enableStoragePools,
 		multiZoneVolumeHandleConfig: multiZoneVolumeHandleConfig,
+		listVolumesConfig:           listVolumesConfig,
 	}
 }
 
diff --git a/pkg/gce-pd-csi-driver/node.go b/pkg/gce-pd-csi-driver/node.go
index 9921a587..b746be75 100644
--- a/pkg/gce-pd-csi-driver/node.go
+++ b/pkg/gce-pd-csi-driver/node.go
@@ -451,10 +451,10 @@ func (ns *GCENodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUns
 		return nil, status.Error(codes.Internal, fmt.Sprintf("NodeUnstageVolume failed: %v\nUnmounting arguments: %s\n", err.Error(), stagingTargetPath))
 	}
 
-	if err := ns.confirmDeviceUnused(volumeID); err != nil {
+	if err := ns.safelyDisableDevice(volumeID); err != nil {
 		var targetErr *ignoreableError
 		if errors.As(err, &targetErr) {
-			klog.Warningf("Unabled to check if device for %s is unused. Device has been unmounted successfully. Ignoring and continuing with unstaging. (%v)", volumeID, err)
+			klog.Warningf("Device may not be detached cleanly for volume %s (ignored, unstaging continues): %v", volumeID, err)
 		} else {
 			return nil, status.Errorf(codes.Internal, "NodeUnstageVolume for volume %s failed: %v", volumeID, err)
 		}
@@ -467,7 +467,7 @@ func (ns *GCENodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUns
 // A private error wrapper used to pass control flow decisions up to the caller
 type ignoreableError struct{ error }
 
-func (ns *GCENodeServer) confirmDeviceUnused(volumeID string) error {
+func (ns *GCENodeServer) safelyDisableDevice(volumeID string) error {
 	devicePath, err := getDevicePath(ns, volumeID, "" /* partition, which is unused */)
 	if err != nil {
 		return &ignoreableError{fmt.Errorf("failed to find device path for volume %s: %v", volumeID, err.Error())}
@@ -484,6 +484,10 @@ func (ns *GCENodeServer) confirmDeviceUnused(volumeID string) error {
 		return fmt.Errorf("device %s (aka %s) is still in use", devicePath, devFsPath)
 	}
 
+	if err := ns.DeviceUtils.DisableDevice(devFsPath); err != nil {
+		return &ignoreableError{fmt.Errorf("failed to disable device %s (aka %s): %v", devicePath, devFsPath, err)}
+	}
+
 	return nil
 }
 


Test patch:
diff --git a/pkg/gce-pd-csi-driver/controller_test.go b/pkg/gce-pd-csi-driver/controller_test.go
index c6fc6a24..17f9946a 100644
--- a/pkg/gce-pd-csi-driver/controller_test.go
+++ b/pkg/gce-pd-csi-driver/controller_test.go
@@ -25,6 +25,8 @@ import (
 	"time"
 
 	"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud/meta"
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
 	compute "google.golang.org/api/compute/v1"
@@ -1133,6 +1135,96 @@ func TestListVolumePagination(t *testing.T) {
 	}
 }
 
+func TestListAttachedVolumePagination(t *testing.T) {
+	testCases := []struct {
+		name            string
+		diskCount       int
+		maxEntries      int32
+		expectedEntries []int
+	}{
+		{
+			name:            "no pagination (implicit)",
+			diskCount:       325,
+			expectedEntries: []int{325},
+		},
+		{
+			name:            "no pagination (explicit)",
+			diskCount:       2500,
+			maxEntries:      2500,
+			expectedEntries: []int{2500},
+		},
+		{
+			name:            "pagination (implicit)",
+			diskCount:       1327,
+			expectedEntries: []int{500, 500, 327},
+		},
+		{
+			name:            "pagination (explicit)",
+			diskCount:       723,
+			maxEntries:      200,
+			expectedEntries: []int{200, 200, 200, 123},
+		},
+		{
+			name:            "pagination (explicit)",
+			diskCount:       3253,
+			maxEntries:      1000,
+			expectedEntries: []int{1000, 1000, 1000, 253},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			// Setup new driver each time so no interference
+			var d []*gce.CloudDisk
+			fakeCloudProvider, err := gce.CreateFakeCloudProvider(project, zone, d)
+			if err != nil {
+				t.Fatalf("Failed to create fake cloud provider: %v", err)
+			}
+			for i := 0; i < tc.diskCount; i++ {
+				// Create diskCount dummy instances, each with a dynamically attached disk
+				instanceName := fmt.Sprintf("instance-%v", i)
+				diskName := fmt.Sprintf("pvc-%v", i)
+				instance := compute.Instance{
+					Name:     name,
+					SelfLink: fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/zones/%s/instances/%s", project, zone, instanceName),
+					Disks: []*compute.AttachedDisk{
+						{
+							DeviceName: diskName,
+							Source:     fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/zones/%s/disks/%s", project, zone, diskName),
+						},
+					},
+				}
+				fakeCloudProvider.InsertInstance(&instance, zone, instanceName)
+			}
+			gceDriver := initGCEDriverWithCloudProvider(t, fakeCloudProvider)
+			// Use attached disks (instances.list) API
+			gceDriver.cs.listVolumesConfig.UseInstancesAPIForPublishedNodes = true
+
+			tok := ""
+			for i, expectedEntry := range tc.expectedEntries {
+				lvr := &csi.ListVolumesRequest{
+					MaxEntries:    tc.maxEntries,
+					StartingToken: tok,
+				}
+				resp, err := gceDriver.cs.ListVolumes(context.TODO(), lvr)
+				if err != nil {
+					t.Fatalf("Got error %v", err)
+					return
+				}
+
+				if len(resp.Entries) != expectedEntry {
+					t.Fatalf("Got %v entries, expected %v on call # %d", len(resp.Entries), expectedEntry, i+1)
+				}
+
+				tok = resp.NextToken
+			}
+			if len(tok) != 0 {
+				t.Fatalf("Expected no more entries, but got NextToken in response: %s", tok)
+			}
+		})
+	}
+}
+
 func TestListVolumeArgs(t *testing.T) {
 	diskCount := 500
 	testCases := []struct {
@@ -1191,6 +1283,204 @@ func TestListVolumeArgs(t *testing.T) {
 	}
 }
 
+func TestListVolumeResponse(t *testing.T) {
+	zone1 := "us-central1-a"
+	zone2 := "us-central1-b"
+	testCases := []struct {
+		name            string
+		disks           []compute.Disk
+		instances       []compute.Instance
+		expectedEntries []*csi.ListVolumesResponse_Entry
+	}{
+		{
+			name: "unattached disk",
+			disks: []compute.Disk{
+				{
+					Name:     "pv-1",
+					SelfLink: fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/zones/%s/disks/%s", project, zone, "pv-1"),
+				},
+			},
+			expectedEntries: []*csi.ListVolumesResponse_Entry{
+				{
+					Volume: &csi.Volume{
+						VolumeId: fmt.Sprintf("projects/%s/zones/%s/disks/%s", project, zone, "pv-1"),
+					},
+					Status: &csi.ListVolumesResponse_VolumeStatus{
+						PublishedNodeIds: []string{},
+					},
+				},
+			},
+		},
+		{
+			name: "attached disk",
+			disks: []compute.Disk{
+				{
+					Name:     "pv-1",
+					SelfLink: fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/zones/%s/disks/%s", project, zone, "pv-1"),
+				},
+			},
+			instances: []compute.Instance{
+				{
+					Name:     "node-1",
+					SelfLink: fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/zones/%s/instances/%s", project, zone, "node-1"),
+					Disks: []*compute.AttachedDisk{
+						{
+							DeviceName: "pv-1",
+							Source:     fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/zones/%s/disks/%s", project, zone, "pv-1"),
+						},
+					},
+				},
+			},
+			expectedEntries: []*csi.ListVolumesResponse_Entry{
+				{
+					Volume: &csi.Volume{
+						VolumeId: fmt.Sprintf("projects/%s/zones/%s/disks/%s", project, zone, "pv-1"),
+					},
+					Status: &csi.ListVolumesResponse_VolumeStatus{
+						PublishedNodeIds: []string{fmt.Sprintf("projects/%s/zones/%s/instances/%s", project, zone, "node-1")},
+					},
+				},
+			},
+		},
+		{
+			name: "attached regional disk",
+			disks: []compute.Disk{
+				{
+					Name:     "pv-1",
+					SelfLink: fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/regions/%s/disks/%s", project, region, "pv-1"),
+				},
+			},
+			instances: []compute.Instance{
+				{
+					Name:     "node-1",
+					SelfLink: fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/zones/%s/instances/%s", project, zone, "node-1"),
+					Disks: []*compute.AttachedDisk{
+						{
+							DeviceName: "pv-1",
+							Source:     fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/regions/%s/disks/%s", project, region, "pv-1"),
+						},
+					},
+				},
+			},
+			expectedEntries: []*csi.ListVolumesResponse_Entry{
+				{
+					Volume: &csi.Volume{
+						VolumeId: fmt.Sprintf("projects/%s/regions/%s/disks/%s", project, region, "pv-1"),
+					},
+					Status: &csi.ListVolumesResponse_VolumeStatus{
+						PublishedNodeIds: []string{fmt.Sprintf("projects/%s/zones/%s/instances/%s", project, zone, "node-1")},
+					},
+				},
+			},
+		},
+		{
+			name: "multi-zone attached disk",
+			disks: []compute.Disk{
+				{
+					Name:     fmt.Sprintf("%s-pv-1", zone1),
+					SelfLink: fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/zones/%s/disks/%s", project, zone1, "pv-1"),
+					Labels:   map[string]string{common.MultiZoneLabel: "true"},
+				},
+				{
+					Name:     fmt.Sprintf("%s-pv-1", zone2),
+					SelfLink: fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/zones/%s/disks/%s", project, zone2, "pv-1"),
+					Labels:   map[string]string{common.MultiZoneLabel: "true"},
+				},
+			},
+			instances: []compute.Instance{
+				{
+					Name:     "node-1",
+					SelfLink: fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/zones/%s/instances/%s", project, zone1, "node-1"),
+					Disks: []*compute.AttachedDisk{
+						{
+							DeviceName: "pv-1",
+							Source:     fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/zones/%s/disks/%s", project, zone1, "pv-1"),
+						},
+					},
+					Zone: zone1,
+				},
+				{
+					Name:     "node-2",
+					SelfLink: fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/zones/%s/instances/%s", project, zone2, "node-2"),
+					Disks: []*compute.AttachedDisk{
+						{
+							DeviceName: "pv-1",
+							Source:     fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/zones/%s/disks/%s", project, zone2, "pv-1"),
+						},
+					},
+					Zone: zone2,
+				},
+			},
+			expectedEntries: []*csi.ListVolumesResponse_Entry{
+				{
+					Volume: &csi.Volume{
+						VolumeId: fmt.Sprintf("projects/%s/zones/%s/disks/%s", project, zone1, "pv-1"),
+					},
+					Status: &csi.ListVolumesResponse_VolumeStatus{
+						PublishedNodeIds: []string{fmt.Sprintf("projects/%s/zones/%s/instances/%s", project, zone1, "node-1")},
+					},
+				},
+				{
+					Volume: &csi.Volume{
+						VolumeId: fmt.Sprintf("projects/%s/zones/%s/disks/%s", project, zone2, "pv-1"),
+					},
+					Status: &csi.ListVolumesResponse_VolumeStatus{
+						PublishedNodeIds: []string{fmt.Sprintf("projects/%s/zones/%s/instances/%s", project, zone2, "node-2")},
+					},
+				},
+				{
+					Volume: &csi.Volume{
+						VolumeId: fmt.Sprintf("projects/%s/zones/multi-zone/disks/%s", project, "pv-1"),
+					},
+					Status: &csi.ListVolumesResponse_VolumeStatus{
+						PublishedNodeIds: []string{
+							fmt.Sprintf("projects/%s/zones/%s/instances/%s", project, zone1, "node-1"),
+							fmt.Sprintf("projects/%s/zones/%s/instances/%s", project, zone2, "node-2"),
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			var d []*gce.CloudDisk
+			for _, disk := range tc.disks {
+				d = append(d, gce.CloudDiskFromV1(&disk))
+			}
+			fakeCloudProvider, err := gce.CreateFakeCloudProvider(project, zone, d)
+			if err != nil {
+			}
+			for _, instance := range tc.instances {
+				fakeCloudProvider.InsertInstance(&instance, instance.Zone, instance.Name)
+			}
+			// Setup new driver each time so no interference
+			gceDriver := initGCEDriverWithCloudProvider(t, fakeCloudProvider)
+			gceDriver.cs.multiZoneVolumeHandleConfig = MultiZoneVolumeHandleConfig{
+				Enable: true,
+			}
+
+			resp, err := gceDriver.cs.ListVolumes(context.TODO(), &csi.ListVolumesRequest{})
+			if err != nil {
+				t.Fatalf("ListVolumes unexpected error: %v", err)
+			}
+
+			if diff := cmp.Diff(tc.expectedEntries, resp.Entries, cmp.Transformer("EntryToVolumeId", entryToVolumeId), cmpopts.SortSlices(less)); diff != "" {
+				t.Errorf("ListVolumes: -want err, +got err\n%s", diff)
+			}
+		})
+	}
+}
+
+func entryToVolumeId(e csi.ListVolumesResponse_Entry) string {
+	return e.Volume.VolumeId
+}
+
+func less(a, b fmt.Stringer) bool {
+	return a.String() < b.String()
+}
+
 func TestCreateVolumeWithVolumeSourceFromSnapshot(t *testing.T) {
 	testCases := []struct {
 		name            string
diff --git a/pkg/gce-pd-csi-driver/gce-pd-driver_test.go b/pkg/gce-pd-csi-driver/gce-pd-driver_test.go
index 0c8a27b3..15306c14 100644
--- a/pkg/gce-pd-csi-driver/gce-pd-driver_test.go
+++ b/pkg/gce-pd-csi-driver/gce-pd-driver_test.go
@@ -49,8 +49,9 @@ func initGCEDriverWithCloudProvider(t *testing.T, cloudProvider gce.GCECompute)
 	fallbackRequisiteZones := []string{}
 	enableStoragePools := false
 	multiZoneVolumeHandleConfig := MultiZoneVolumeHandleConfig{}
+	listVolumesConfig := ListVolumesConfig{}
 
-	controllerServer := NewControllerServer(gceDriver, cloudProvider, errorBackoffInitialDuration, errorBackoffMaxDuration, fallbackRequisiteZones, enableStoragePools, multiZoneVolumeHandleConfig)
+	controllerServer := NewControllerServer(gceDriver, cloudProvider, errorBackoffInitialDuration, errorBackoffMaxDuration, fallbackRequisiteZones, enableStoragePools, multiZoneVolumeHandleConfig, listVolumesConfig)
 	err := gceDriver.SetupGCEDriver(driver, vendorVersion, nil, nil, nil, controllerServer, nil)
 	if err != nil {
 		t.Fatalf("Failed to setup GCE Driver: %v", err)
diff --git a/test/e2e/utils/utils.go b/test/e2e/utils/utils.go
index 00e8bb65..a044b689 100644
--- a/test/e2e/utils/utils.go
+++ b/test/e2e/utils/utils.go
@@ -56,13 +56,17 @@ func GCEClientAndDriverSetup(instance *remote.InstanceInfo, computeEndpoint stri
 	extra_flags := []string{
 		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",
+		"--use-instance-api-to-poll-attachment-disk-types=pd-ssd",
+		"--use-instance-api-to-list-volumes-published-nodes",
 	}
 	extra_flags = append(extra_flags, fmt.Sprintf("--compute-endpoint=%s", computeEndpoint))
 
 	workspace := remote.NewWorkspaceDir("gce-pd-e2e-")
 	// Log at V(6) as the compute API calls are emitted at that level and it's
 	// useful to see what's happening when debugging tests.
-	driverRunCmd := fmt.Sprintf("sh -c '/usr/bin/nohup %s/gce-pd-csi-driver -v=6 --endpoint=%s --multi-zone-volume-handle-enable --multi-zone-volume-handle-disk-types=pd-standard --use-instance-api-to-poll-attachment-disk-types=pd-ssd %s 2> %s/prog.out < /dev/null > /dev/null &'",
+	driverRunCmd := fmt.Sprintf("sh -c '/usr/bin/nohup %s/gce-pd-csi-driver -v=6 --endpoint=%s %s 2> %s/prog.out < /dev/null > /dev/null &'",
 		workspace, endpoint, strings.Join(extra_flags, " "), workspace)
 	config := &remote.ClientConfig{
 		PkgPath:      pkgPath,
diff --git a/test/sanity/sanity_test.go b/test/sanity/sanity_test.go
index 890884fc..f09c64ef 100644
--- a/test/sanity/sanity_test.go
+++ b/test/sanity/sanity_test.go
@@ -64,13 +64,14 @@ func TestSanity(t *testing.T) {
 	fallbackRequisiteZones := []string{}
 	enableStoragePools := false
 	multiZoneVolumeHandleConfig := driver.MultiZoneVolumeHandleConfig{}
+	listVolumesConfig := driver.ListVolumesConfig{}
 
 	mounter := mountmanager.NewFakeSafeMounter()
 	deviceUtils := deviceutils.NewFakeDeviceUtils(true)
 
 	//Initialize GCE Driver
 	identityServer := driver.NewIdentityServer(gceDriver)
-	controllerServer := driver.NewControllerServer(gceDriver, cloudProvider, 0, 5*time.Minute, fallbackRequisiteZones, enableStoragePools, multiZoneVolumeHandleConfig)
+	controllerServer := driver.NewControllerServer(gceDriver, cloudProvider, 0, 5*time.Minute, fallbackRequisiteZones, enableStoragePools, multiZoneVolumeHandleConfig, listVolumesConfig)
 	fakeStatter := mountmanager.NewFakeStatterWithOptions(mounter, mountmanager.FakeStatterOptions{IsBlock: false})
 	nodeServer := driver.NewNodeServer(gceDriver, mounter, deviceUtils, metadataservice.NewFakeService(), fakeStatter)
 	err = gceDriver.SetupGCEDriver(driverName, vendorVersion, extraLabels, nil, identityServer, controllerServer, nodeServer)