Repository: kubernetes-sigs/gcp-compute-persistent-disk-csi-driver
Pull request #923 (2022-02-16 22:10:07)

Ensure max one running attach/detach operation for a given disk on an instance
<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our contributor guidelines: https://git.k8s.io/community/contributors/guide#your-first-contribution and developer guide https://git.k8s.io/community/contributors/devel/development.md#development-guide
2. Follow the instructions for writing a release note: https://git.k8s.io/community/contributors/guide/release-notes.md
3. If the PR is unfinished, see how to mark it: https://git.k8s.io/community/contributors/guide/pull-requests.md#marking-unfinished-pull-requests
-->

**What type of PR is this?**
> Uncomment only one ` /kind <>` line, hit enter to put that in a new line, and remove leading whitespaces from that line:
>
> /kind api-change
> /kind bug
> /kind cleanup
> /kind design
> /kind documentation
> /kind failing-test
/kind feature
> /kind flake

**What this PR does / why we need it**:
This PR changes the queuing behavior of ControllerPublish/Unpublish. The controller will try to ensure at max 1 attach or detach running operation for a given PD on a given instance. 
The Controller maintains two in-memory maps 
1. last known running attach detach ops for a disk on a given instance
2. last known running ops for an instance  (this map is needed to handle the driver restart scenario)

These maps are checked before initiating a new controller publish/unpublish call.

For detailed design check [doc](https://docs.google.com/document/d/1WIbf0QsjReGBJbrSVWp90TxnRgbP3bU01Bj3KIDCV3g/edit?usp=sharing)

The PR :
1. Basic caching behavior implementation, ops manager to handle ops cache consistency, safe start of attach detach operations.
2. Cache hydration logic.

**Which issue(s) this PR fixes**:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes #

**Special notes for your reviewer**:

**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
Ensure max one running attach/detach operation for a given disk on an instance
```


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/923#" title="Author self-approved">saikat-royc</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fgcp-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/release-999.0/OWNERS)~~ [saikat-royc]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
k8s-ci-robot: @saikat-royc: GitHub didn't allow me to request PR reviews from the following users: pwschuurman.

Note that only [kubernetes-sigs members](https://github.com/orgs/kubernetes-sigs/people) and repo collaborators can review this PR, and authors cannot review their own PRs.

<details>

In response to [this](https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/pull/923#issuecomment-1042362108):

>/cc pwschuurman
>


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
saikat-royc: /cc pwschuurman

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/923#pullrequestreview-895657305" title="Approved">pwschuurman</a>*, *<a href="https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/pull/923#" title="Author self-approved">saikat-royc</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fgcp-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/release-999.0/OWNERS)~~ [saikat-royc]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->

Patch:
diff --git a/pkg/common/cache.go b/pkg/common/cache.go
new file mode 100644
index 00000000..c5e31ce3
--- /dev/null
+++ b/pkg/common/cache.go
@@ -0,0 +1,140 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package common
+
+import (
+	"fmt"
+)
+
+type DiskInstanceKey string
+type InstanceKey string
+
+type OpInfo struct {
+	Name string
+	Type string
+}
+
+type DiskInstanceOpsMap struct {
+	ops map[DiskInstanceKey]OpInfo
+}
+
+func NewDiskInstanceOpsMap() *DiskInstanceOpsMap {
+	return &DiskInstanceOpsMap{
+		ops: make(map[DiskInstanceKey]OpInfo),
+	}
+}
+
+func (c *DiskInstanceOpsMap) GetOp(key DiskInstanceKey) *OpInfo {
+	op, ok := c.ops[key]
+	if !ok {
+		return nil
+	}
+	return &op
+}
+
+func (c *DiskInstanceOpsMap) ClearOp(key DiskInstanceKey, targetOpName string) error {
+	op, ok := c.ops[key]
+	if !ok {
+		return nil
+	}
+
+	if op.Name != targetOpName {
+		return fmt.Errorf("For key %q, cannot clear op %q, cache already contains op %q", key, targetOpName, op.Name)
+	}
+
+	delete(c.ops, key)
+	return nil
+}
+
+func (c *DiskInstanceOpsMap) AddOp(key DiskInstanceKey, targetOp OpInfo) {
+	c.ops[key] = targetOp
+}
+
+type InstanceOpsMap struct {
+	ops map[InstanceKey][]OpInfo
+}
+
+func NewInstanceOpsMap() *InstanceOpsMap {
+	return &InstanceOpsMap{
+		ops: make(map[InstanceKey][]OpInfo),
+	}
+}
+
+func (c *InstanceOpsMap) GetOps(key InstanceKey) []OpInfo {
+	v, ok := c.ops[key]
+	if !ok {
+		return nil
+	}
+	return v
+}
+
+func (c *InstanceOpsMap) ClearOp(key InstanceKey, targetOpName string) {
+	ops, ok := c.ops[key]
+	if !ok {
+		return
+	}
+
+	var filteredOps []OpInfo
+	for _, o := range ops {
+		if o.Name != targetOpName {
+			filteredOps = append(filteredOps, o)
+		}
+	}
+	if len(filteredOps) == 0 {
+		delete(c.ops, key)
+		return
+	}
+
+	c.ops[key] = filteredOps
+}
+
+func (c *InstanceOpsMap) AddOp(key InstanceKey, targetOp OpInfo) {
+	ops, ok := c.ops[key]
+	if !ok {
+		c.ops[key] = []OpInfo{targetOp}
+		return
+	}
+
+	for _, o := range ops {
+		if o.Name == targetOp.Name {
+			return
+		}
+	}
+
+	ops = append(ops, targetOp)
+	c.ops[key] = ops
+}
+
+type OpsCache struct {
+	InstanceOps     *InstanceOpsMap
+	DiskInstanceOps *DiskInstanceOpsMap
+}
+
+func NewOpsCache() *OpsCache {
+	return &OpsCache{
+		InstanceOps:     NewInstanceOpsMap(),
+		DiskInstanceOps: NewDiskInstanceOpsMap(),
+	}
+}
+
+func CreateDiskInstanceKey(project, location, disk, instance string) DiskInstanceKey {
+	return DiskInstanceKey(fmt.Sprintf("%s_%s_%s_%s", project, location, disk, instance))
+}
+
+func CreateInstanceKey(project, location, instance string) InstanceKey {
+	return InstanceKey(fmt.Sprintf("%s_%s_%s", project, location, instance))
+}
diff --git a/pkg/common/utils.go b/pkg/common/utils.go
index 6b71feb7..87d9ac6d 100644
--- a/pkg/common/utils.go
+++ b/pkg/common/utils.go
@@ -238,3 +238,24 @@ func ProcessStorageLocations(storageLocations string) ([]string, error) {
 	}
 	return []string{normalizedLoc}, nil
 }
+
+func ParseOpTargetLinkUrl(url string) (string, string, string, error) {
+	if url == "" {
+		return "", "", "", fmt.Errorf("url is empty")
+	}
+
+	// example url: https://www.googleapis.com/compute/v1/projects/<projectname>/zones/<location>/instances/<instance-name>
+	s := strings.TrimPrefix(url, "https://www.googleapis.com/compute/")
+	splitStr := strings.Split(s, "/")
+	if len(splitStr) != 7 {
+		return "", "", "", fmt.Errorf("unknown targetLink format %s", url)
+	}
+
+	project := splitStr[2]
+	location := splitStr[4]
+	instanceName := splitStr[6]
+	if project == "" || location == "" || instanceName == "" {
+		return "", "", "", fmt.Errorf("unknown targetLink format %s", url)
+	}
+	return project, location, instanceName, nil
+}
diff --git a/pkg/gce-cloud-provider/compute/fake-gce.go b/pkg/gce-cloud-provider/compute/fake-gce.go
index 6a7adeff..876eba31 100644
--- a/pkg/gce-cloud-provider/compute/fake-gce.go
+++ b/pkg/gce-cloud-provider/compute/fake-gce.go
@@ -22,6 +22,7 @@ import (
 
 	"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud/meta"
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/google/uuid"
 	computev1 "google.golang.org/api/compute/v1"
 	"google.golang.org/api/googleapi"
 	"google.golang.org/grpc/codes"
@@ -47,6 +48,7 @@ type FakeCloudProvider struct {
 	pageTokens map[string]sets.String
 	instances  map[string]*computev1.Instance
 	snapshots  map[string]*computev1.Snapshot
+	ops        []*computev1.Operation
 
 	// marker to set disk status during InsertDisk operation.
 	mockDiskStatus string
@@ -266,41 +268,6 @@ func (cloud *FakeCloudProvider) DeleteDisk(ctx context.Context, project string,
 	return nil
 }
 
-func (cloud *FakeCloudProvider) AttachDisk(ctx context.Context, project string, volKey *meta.Key, readWrite, diskType, instanceZone, instanceName string) error {
-	source := cloud.GetDiskSourceURI(project, volKey)
-
-	attachedDiskV1 := &computev1.AttachedDisk{
-		DeviceName: volKey.Name,
-		Kind:       diskKind,
-		Mode:       readWrite,
-		Source:     source,
-		Type:       diskType,
-	}
-	instance, ok := cloud.instances[instanceName]
-	if !ok {
-		return fmt.Errorf("Failed to get instance %v", instanceName)
-	}
-	instance.Disks = append(instance.Disks, attachedDiskV1)
-	return nil
-}
-
-func (cloud *FakeCloudProvider) DetachDisk(ctx context.Context, project, deviceName, instanceZone, instanceName string) error {
-	instance, ok := cloud.instances[instanceName]
-	if !ok {
-		return fmt.Errorf("Failed to get instance %v", instanceName)
-	}
-	found := -1
-	for i, disk := range instance.Disks {
-		if disk.DeviceName == deviceName {
-			found = i
-			break
-		}
-	}
-	instance.Disks[found] = instance.Disks[len(instance.Disks)-1]
-	instance.Disks = instance.Disks[:len(instance.Disks)-1]
-	return nil
-}
-
 func (cloud *FakeCloudProvider) GetDiskTypeURI(project string, volKey *meta.Key, diskType string) string {
 	switch volKey.Type() {
 	case meta.Zonal:
@@ -451,9 +418,14 @@ func (cloud *FakeCloudProvider) UpdateDiskStatus(s string) {
 	cloud.mockDiskStatus = s
 }
 
+type Signal struct {
+	ReportError   bool
+	ReportRunning bool
+}
+
 type FakeBlockingCloudProvider struct {
 	*FakeCloudProvider
-	ReadyToExecute chan chan struct{}
+	ReadyToExecute chan chan Signal
 }
 
 // FakeBlockingCloudProvider's method adds functionality to finely control the order of execution of CreateSnapshot calls.
@@ -461,12 +433,36 @@ type FakeBlockingCloudProvider struct {
 // The test calling this function can block on readyToExecute to ensure that the operation has started and
 // allowed the CreateSnapshot to continue by passing a struct into executeCreateSnapshot.
 func (cloud *FakeBlockingCloudProvider) CreateSnapshot(ctx context.Context, project string, volKey *meta.Key, snapshotName string, snapshotParams common.SnapshotParameters) (*computev1.Snapshot, error) {
-	executeCreateSnapshot := make(chan struct{})
+	executeCreateSnapshot := make(chan Signal)
 	cloud.ReadyToExecute <- executeCreateSnapshot
 	<-executeCreateSnapshot
 	return cloud.FakeCloudProvider.CreateSnapshot(ctx, project, volKey, snapshotName, snapshotParams)
 }
 
+func (cloud *FakeBlockingCloudProvider) WaitForZonalOp(ctx context.Context, project, opName string, zone string) error {
+	execute := make(chan Signal)
+	cloud.ReadyToExecute <- execute
+	val := <-execute
+	if val.ReportError {
+		return fmt.Errorf("force mock error of zonal op %s", opName)
+	}
+	return nil
+}
+
+func (cloud *FakeBlockingCloudProvider) CheckZonalOpDoneStatus(ctx context.Context, project, location, opId string) (bool, error) {
+	execute := make(chan Signal)
+	cloud.ReadyToExecute <- execute
+	val := <-execute
+	if val.ReportError {
+		return false, fmt.Errorf("force mock error of zonal op %s", opId)
+	}
+
+	if val.ReportRunning {
+		return false, nil
+	}
+	return cloud.FakeCloudProvider.CheckZonalOpDoneStatus(ctx, project, location, opId)
+}
+
 func notFoundError() *googleapi.Error {
 	return &googleapi.Error{
 		Errors: []googleapi.ErrorItem{
@@ -486,3 +482,67 @@ func invalidError() *googleapi.Error {
 		},
 	}
 }
+
+func (cloud *FakeCloudProvider) StartAttachDiskOp(ctx context.Context, volKey *meta.Key, readWrite, diskType, project, location, instanceName string) (*computev1.Operation, error) {
+	source := cloud.GetDiskSourceURI(project, volKey)
+	attachedDiskV1 := &computev1.AttachedDisk{
+		DeviceName: volKey.Name,
+		Kind:       diskKind,
+		Mode:       readWrite,
+		Source:     source,
+		Type:       diskType,
+	}
+	instance, ok := cloud.instances[instanceName]
+	if !ok {
+		return nil, fmt.Errorf("Failed to get instance %v", instanceName)
+	}
+	instance.Disks = append(instance.Disks, attachedDiskV1)
+	op := &computev1.Operation{Name: "op" + uuid.New().String(), OperationType: "attachDisk"}
+	cloud.ops = append(cloud.ops, op)
+	return op, nil
+}
+
+func (cloud *FakeCloudProvider) StartDetachDiskOp(ctx context.Context, project, location, deviceName, instanceName string) (*computev1.Operation, error) {
+	instance, ok := cloud.instances[instanceName]
+	if !ok {
+		return nil, fmt.Errorf("Failed to get instance %v", instanceName)
+	}
+	found := -1
+	for i, disk := range instance.Disks {
+		if disk.DeviceName == deviceName {
+			found = i
+			break
+		}
+	}
+	instance.Disks[found] = instance.Disks[len(instance.Disks)-1]
+	instance.Disks = instance.Disks[:len(instance.Disks)-1]
+	op := &computev1.Operation{Name: "op" + uuid.New().String(), OperationType: "detachDisk"}
+	cloud.ops = append(cloud.ops, op)
+	return op, nil
+}
+
+func (cloud *FakeCloudProvider) OpIsDone(op *computev1.Operation) (bool, error) {
+	if op == nil || op.Status != operationStatusDone {
+		return false, nil
+	}
+	if op.Error != nil && len(op.Error.Errors) > 0 && op.Error.Errors[0] != nil {
+		return true, fmt.Errorf("operation %v failed (%v): %v", op.Name, op.Error.Errors[0].Code, op.Error.Errors[0].Message)
+	}
+	return true, nil
+}
+
+func (cloud *FakeCloudProvider) CheckZonalOpDoneStatus(ctx context.Context, project, location, opId string) (bool, error) {
+	return true, nil
+}
+
+func (cloud *FakeCloudProvider) WaitForZonalOp(ctx context.Context, project, opName string, zone string) error {
+	return nil
+}
+
+func (cloud *FakeCloudProvider) ListZonalOps(ctx context.Context, opTypeFilter map[string]bool) ([]*computev1.Operation, error) {
+	return cloud.ops, nil
+}
+
+func (cloud *FakeCloudProvider) InsertOps(ops []*computev1.Operation) {
+	cloud.ops = ops
+}
diff --git a/pkg/gce-cloud-provider/compute/gce-compute.go b/pkg/gce-cloud-provider/compute/gce-compute.go
index f8949950..1f604da3 100644
--- a/pkg/gce-cloud-provider/compute/gce-compute.go
+++ b/pkg/gce-cloud-provider/compute/gce-compute.go
@@ -58,8 +58,8 @@ type GCECompute interface {
 	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
 	DeleteDisk(ctx context.Context, project string, volumeKey *meta.Key) error
-	AttachDisk(ctx context.Context, project string, volKey *meta.Key, readWrite, diskType, instanceZone, instanceName string) error
-	DetachDisk(ctx context.Context, project, deviceName, instanceZone, instanceName string) error
+	StartAttachDiskOp(ctx context.Context, volKey *meta.Key, readWrite, diskType, project, location, instanceName string) (*computev1.Operation, error)
+	StartDetachDiskOp(ctx context.Context, project, location, deviceName, instanceName string) (*computev1.Operation, 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, instanceZone, instanceName string) error
@@ -75,6 +75,13 @@ type GCECompute interface {
 	GetSnapshot(ctx context.Context, project, snapshotName string) (*computev1.Snapshot, error)
 	CreateSnapshot(ctx context.Context, project string, volKey *meta.Key, snapshotName string, snapshotParams common.SnapshotParameters) (*computev1.Snapshot, error)
 	DeleteSnapshot(ctx context.Context, project, snapshotName string) error
+	// Operation methods
+	// Evaluate the status on a operation object.
+	OpIsDone(op *computev1.Operation) (bool, error)
+	// Perform GET of operation and then evaluate the status.
+	CheckZonalOpDoneStatus(ctx context.Context, project, location, opId string) (bool, error)
+	WaitForZonalOp(ctx context.Context, project, opName string, zone string) error
+	ListZonalOps(ctx context.Context, opTypeFilter map[string]bool) ([]*computev1.Operation, error)
 }
 
 // GetDefaultProject returns the project that was used to instantiate this GCE client.
@@ -547,7 +554,7 @@ func (cloud *CloudProvider) insertZonalDisk(
 	}
 	klog.V(5).Infof("InsertDisk operation %s for disk %s", opName, diskToCreate.Name)
 
-	err = cloud.waitForZonalOp(ctx, project, opName, volKey.Zone)
+	err = cloud.WaitForZonalOp(ctx, project, opName, volKey.Zone)
 
 	if err != nil {
 		if IsGCEError(err, "alreadyExists") {
@@ -593,7 +600,7 @@ func (cloud *CloudProvider) deleteZonalDisk(ctx context.Context, project, zone,
 	}
 	klog.V(5).Infof("DeleteDisk operation %s for disk %s", op.Name, name)
 
-	err = cloud.waitForZonalOp(ctx, project, op.Name, zone)
+	err = cloud.WaitForZonalOp(ctx, project, op.Name, zone)
 	if err != nil {
 		return err
 	}
@@ -618,50 +625,6 @@ func (cloud *CloudProvider) deleteRegionalDisk(ctx context.Context, project, reg
 	return nil
 }
 
-func (cloud *CloudProvider) AttachDisk(ctx context.Context, project string, volKey *meta.Key, readWrite, diskType, instanceZone, instanceName string) error {
-	klog.V(5).Infof("Attaching disk %v to %s", volKey, instanceName)
-	source := cloud.GetDiskSourceURI(project, volKey)
-
-	deviceName, err := common.GetDeviceName(volKey)
-	if err != nil {
-		return fmt.Errorf("failed to get device name: %v", err)
-	}
-	attachedDiskV1 := &computev1.AttachedDisk{
-		DeviceName: deviceName,
-		Kind:       diskKind,
-		Mode:       readWrite,
-		Source:     source,
-		Type:       diskType,
-	}
-
-	op, err := cloud.service.Instances.AttachDisk(project, instanceZone, instanceName, attachedDiskV1).Context(ctx).Do()
-	if err != nil {
-		return fmt.Errorf("failed cloud service attach disk call: %v", err)
-	}
-	klog.V(5).Infof("AttachDisk operation %s for disk %s", op.Name, attachedDiskV1.DeviceName)
-
-	err = cloud.waitForZonalOp(ctx, project, op.Name, instanceZone)
-	if err != nil {
-		return fmt.Errorf("failed when waiting for zonal op: %v", err)
-	}
-	return nil
-}
-
-func (cloud *CloudProvider) DetachDisk(ctx context.Context, project, deviceName, instanceZone, instanceName string) error {
-	klog.V(5).Infof("Detaching disk %v from %v", deviceName, instanceName)
-	op, err := cloud.service.Instances.DetachDisk(project, instanceZone, instanceName, deviceName).Context(ctx).Do()
-	if err != nil {
-		return err
-	}
-	klog.V(5).Infof("DetachDisk operation %s for disk %s", op.Name, deviceName)
-
-	err = cloud.waitForZonalOp(ctx, project, op.Name, instanceZone)
-	if err != nil {
-		return err
-	}
-	return nil
-}
-
 func (cloud *CloudProvider) GetDiskSourceURI(project string, volKey *meta.Key) string {
 	switch volKey.Type() {
 	case meta.Zonal:
@@ -708,7 +671,7 @@ func (cloud *CloudProvider) getRegionalDiskTypeURI(project string, region, diskT
 	return cloud.service.BasePath + fmt.Sprintf(diskTypeURITemplateRegional, project, region, diskType)
 }
 
-func (cloud *CloudProvider) waitForZonalOp(ctx context.Context, project, opName string, zone string) error {
+func (cloud *CloudProvider) WaitForZonalOp(ctx context.Context, project, opName string, zone string) error {
 	// The v1 API can query for v1, alpha, or beta operations.
 	return wait.Poll(3*time.Second, 5*time.Minute, func() (bool, error) {
 		pollOp, err := cloud.service.ZoneOperations.Get(project, zone, opName).Context(ctx).Do()
@@ -716,7 +679,7 @@ func (cloud *CloudProvider) waitForZonalOp(ctx context.Context, project, opName
 			klog.Errorf("WaitForOp(op: %s, zone: %#v) failed to poll the operation", opName, zone)
 			return false, err
 		}
-		done, err := opIsDone(pollOp)
+		done, err := cloud.OpIsDone(pollOp)
 		return done, err
 	})
 }
@@ -729,7 +692,7 @@ func (cloud *CloudProvider) waitForRegionalOp(ctx context.Context, project, opNa
 			klog.Errorf("WaitForOp(op: %s, region: %#v) failed to poll the operation", opName, region)
 			return false, err
 		}
-		done, err := opIsDone(pollOp)
+		done, err := cloud.OpIsDone(pollOp)
 		return done, err
 	})
 }
@@ -741,7 +704,7 @@ func (cloud *CloudProvider) waitForGlobalOp(ctx context.Context, project, opName
 			klog.Errorf("waitForGlobalOp(op: %s) failed to poll the operation", opName)
 			return false, err
 		}
-		done, err := opIsDone(pollOp)
+		done, err := cloud.OpIsDone(pollOp)
 		return done, err
 	})
 }
@@ -769,8 +732,11 @@ func (cloud *CloudProvider) WaitForAttach(ctx context.Context, project string, v
 	})
 }
 
-func opIsDone(op *computev1.Operation) (bool, error) {
-	if op == nil || op.Status != operationStatusDone {
+func (cloud *CloudProvider) OpIsDone(op *computev1.Operation) (bool, error) {
+	if op == nil {
+		return true, nil
+	}
+	if op.Status != operationStatusDone {
 		return false, nil
 	}
 	if op.Error != nil && len(op.Error.Errors) > 0 && op.Error.Errors[0] != nil {
@@ -869,7 +835,7 @@ func (cloud *CloudProvider) resizeZonalDisk(ctx context.Context, project string,
 	}
 	klog.V(5).Infof("ResizeDisk operation %s for disk %s", op.Name, volKey.Name)
 
-	err = cloud.waitForZonalOp(ctx, project, op.Name, volKey.Zone)
+	err = cloud.WaitForZonalOp(ctx, project, op.Name, volKey.Zone)
 	if err != nil {
 		return -1, fmt.Errorf("failed waiting for op for zonal resize for %s: %v", volKey.String(), err)
 	}
@@ -953,6 +919,86 @@ func (cloud *CloudProvider) waitForSnapshotCreation(ctx context.Context, project
 	}
 }
 
+func (cloud *CloudProvider) StartAttachDiskOp(ctx context.Context, volKey *meta.Key, readWrite, diskType, project, location, instanceName string) (*computev1.Operation, error) {
+	deviceName, err := common.GetDeviceName(volKey)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get device name: %v", err)
+	}
+
+	attachedDiskV1 := &computev1.AttachedDisk{
+		DeviceName: deviceName,
+		Kind:       diskKind,
+		Mode:       readWrite,
+		Source:     cloud.GetDiskSourceURI(project, volKey),
+		Type:       diskType,
+	}
+
+	op, err := cloud.service.Instances.AttachDisk(project, location, instanceName, attachedDiskV1).Context(ctx).Do()
+	if err != nil {
+		return nil, fmt.Errorf("failed cloud service attach disk call: %v", err)
+	}
+	return op, nil
+}
+
+func (cloud *CloudProvider) StartDetachDiskOp(ctx context.Context, project, location, deviceName, instanceName string) (*computev1.Operation, error) {
+	op, err := cloud.service.Instances.DetachDisk(project, location, instanceName, deviceName).Context(ctx).Do()
+	if err != nil {
+		return nil, fmt.Errorf("failed cloud service detach disk call: %v", err)
+	}
+	return op, nil
+}
+
+// CheckZonalOpDoneStatus returns true, if the op is complete. Returns false otherwise. Returns an error if GET returns an error.
+func (cloud *CloudProvider) CheckZonalOpDoneStatus(ctx context.Context, project, location, opId string) (bool, error) {
+	lastKnownOp, err := cloud.service.ZoneOperations.Get(project, location, opId).Context(ctx).Do()
+	if err != nil {
+		if !IsGCENotFoundError(err) {
+			return false, fmt.Errorf("failed to get operation %s: %v", opId, err)
+		}
+		return true, nil
+	}
+
+	// If the op complete with errors, the op is still complete.
+	if done, _ := cloud.OpIsDone(lastKnownOp); done {
+		return true, nil
+	}
+
+	return false, nil
+}
+
+func (cloud *CloudProvider) ListZonalOps(ctx context.Context, opTypeFilter map[string]bool) ([]*computev1.Operation, error) {
+	region, err := common.GetRegionFromZones([]string{cloud.zone})
+	if err != nil {
+		return nil, fmt.Errorf("failed to get region from zones: %v", err)
+	}
+	zones, err := cloud.ListZones(ctx, region)
+	if err != nil {
+		return nil, err
+	}
+
+	items := []*computev1.Operation{}
+	for _, zone := range zones {
+		lCall := cloud.service.ZoneOperations.List(cloud.project, zone)
+		nextPageToken := "pageToken"
+		for nextPageToken != "" {
+			opList, err := lCall.Do()
+			if err != nil {
+				return nil, err
+			}
+
+			for _, op := range opList.Items {
+				if _, ok := opTypeFilter[op.OperationType]; ok {
+					items = append(items, op)
+				}
+			}
+
+			nextPageToken = opList.NextPageToken
+			lCall.PageToken(nextPageToken)
+		}
+	}
+	return items, nil
+}
+
 // kmsKeyEqual returns true if fetchedKMSKey and storageClassKMSKey refer to the same key.
 // fetchedKMSKey - key returned by the server
 //        example: projects/{0}/locations/{1}/keyRings/{2}/cryptoKeys/{3}/cryptoKeyVersions/{4}
diff --git a/pkg/gce-cloud-provider/compute/gce.go b/pkg/gce-cloud-provider/compute/gce.go
index 2385889b..c699c913 100644
--- a/pkg/gce-cloud-provider/compute/gce.go
+++ b/pkg/gce-cloud-provider/compute/gce.go
@@ -56,8 +56,7 @@ type CloudProvider struct {
 	betaService *computebeta.Service
 	project     string
 	zone        string
-
-	zonesCache map[string][]string
+	zonesCache  map[string][]string
 }
 
 var _ GCECompute = &CloudProvider{}
diff --git a/pkg/gce-pd-csi-driver/controller.go b/pkg/gce-pd-csi-driver/controller.go
index 2ffa96e8..595baf9e 100644
--- a/pkg/gce-pd-csi-driver/controller.go
+++ b/pkg/gce-pd-csi-driver/controller.go
@@ -59,6 +59,8 @@ type GCEControllerServer struct {
 	// the attach/detach operations until there is an attach / detach
 	// operation succeeds
 	publishErrorsSeenOnNode map[string]bool
+
+	opsManager *OpsManager
 }
 
 type workItem struct {
@@ -334,6 +336,7 @@ func (gceCS *GCEControllerServer) DeleteVolume(ctx context.Context, req *csi.Del
 
 // Run starts the GCEControllerServer.
 func (gceCS *GCEControllerServer) Run() {
+	go gceCS.opsManager.HydrateOpsCache()
 	go wait.Until(gceCS.worker, 1*time.Second, wait.NeverStop)
 }
 
@@ -377,6 +380,10 @@ func (gceCS *GCEControllerServer) processNextWorkItem() bool {
 }
 
 func (gceCS *GCEControllerServer) ControllerPublishVolume(ctx context.Context, req *csi.ControllerPublishVolumeRequest) (*csi.ControllerPublishVolumeResponse, error) {
+	if !gceCS.opsManager.IsReady() {
+		return nil, status.Errorf(codes.Aborted, "Cache not ready")
+	}
+
 	// Only valid requests will be queued
 	_, _, err := gceCS.validateControllerPublishVolumeRequest(ctx, req)
 
@@ -495,13 +502,24 @@ 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
 	}
+
+	// Check and initiate an attach disk operation.
 	instanceZone, instanceName, err = common.NodeIDToZoneAndName(nodeID)
 	if err != nil {
 		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("could not split nodeID: %v", err))
 	}
-	err = gceCS.CloudProvider.AttachDisk(ctx, project, volKey, readWrite, attachableDiskTypePersistent, instanceZone, instanceName)
+
+	err = gceCS.opsManager.ExecuteAttachDisk(ctx, &AttachDiskOpts{
+		Volumekey:    volKey,
+		ReadWrite:    readWrite,
+		DiskType:     attachableDiskTypePersistent,
+		Project:      project,
+		Location:     instanceZone,
+		DeviceName:   deviceName,
+		InstanceName: instanceName,
+	})
 	if err != nil {
-		return nil, status.Error(codes.Internal, fmt.Sprintf("unknown Attach error: %v", err))
+		return nil, err
 	}
 
 	err = gceCS.CloudProvider.WaitForAttach(ctx, project, volKey, instanceZone, instanceName)
@@ -520,6 +538,10 @@ func (gceCS *GCEControllerServer) executeControllerPublishVolume(ctx context.Con
 }
 
 func (gceCS *GCEControllerServer) ControllerUnpublishVolume(ctx context.Context, req *csi.ControllerUnpublishVolumeRequest) (*csi.ControllerUnpublishVolumeResponse, error) {
+	if !gceCS.opsManager.IsReady() {
+		return nil, status.Errorf(codes.Aborted, "Cache not ready")
+	}
+
 	// Only valid requests will be queued
 	_, _, err := gceCS.validateControllerUnpublishVolumeRequest(ctx, req)
 
@@ -612,17 +634,17 @@ func (gceCS *GCEControllerServer) executeControllerUnpublishVolume(ctx context.C
 		return &csi.ControllerUnpublishVolumeResponse{}, nil
 	}
 
-	err = gceCS.CloudProvider.DetachDisk(ctx, project, deviceName, instanceZone, instanceName)
+	err = gceCS.opsManager.ExecuteDetachDisk(ctx, &DetachDiskOpts{
+		Project:      project,
+		Location:     instanceZone,
+		DeviceName:   deviceName,
+		InstanceName: instanceName,
+	})
 	if err != nil {
-		// Mark the node and rate limit all the following attach/detach
-		// operations for this node
-		gceCS.publishErrorsSeenOnNode[nodeID] = true
-		return nil, status.Error(codes.Internal, fmt.Sprintf("unknown detach error: %v", err))
+		return nil, err
 	}
 
-	// Detach succeeds so unmark the node
 	delete(gceCS.publishErrorsSeenOnNode, nodeID)
-
 	klog.V(4).Infof("ControllerUnpublishVolume succeeded for disk %v from node %v", volKey, nodeID)
 	return &csi.ControllerUnpublishVolumeResponse{}, nil
 }
diff --git a/pkg/gce-pd-csi-driver/gce-pd-driver.go b/pkg/gce-pd-csi-driver/gce-pd-driver.go
index 17d644ef..67054ff3 100644
--- a/pkg/gce-pd-csi-driver/gce-pd-driver.go
+++ b/pkg/gce-pd-csi-driver/gce-pd-driver.go
@@ -157,6 +157,7 @@ func NewControllerServer(gceDriver *GCEDriver, cloudProvider gce.GCECompute) *GC
 		volumeLocks:             common.NewVolumeLocks(),
 		queue:                   workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "controllerserver"),
 		publishErrorsSeenOnNode: map[string]bool{},
+		opsManager:              NewOpsManager(cloudProvider),
 	}
 }
 
diff --git a/pkg/gce-pd-csi-driver/ops_manager.go b/pkg/gce-pd-csi-driver/ops_manager.go
new file mode 100644
index 00000000..1e2eebcb
--- /dev/null
+++ b/pkg/gce-pd-csi-driver/ops_manager.go
@@ -0,0 +1,243 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package gceGCEDriver
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud/meta"
+	compute "google.golang.org/api/compute/v1"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/util/retry"
+	"k8s.io/klog"
+	"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"
+)
+
+var (
+	backoff = wait.Backoff{
+		Steps:    300,
+		Duration: 1 * time.Second,
+		Factor:   1.0,
+	}
+)
+
+type AttachDiskOpts struct {
+	Volumekey    *meta.Key
+	ReadWrite    string
+	DiskType     string
+	Project      string
+	Location     string
+	DeviceName   string
+	InstanceName string
+}
+
+type DetachDiskOpts struct {
+	Project      string
+	Location     string
+	DeviceName   string
+	InstanceName string
+}
+
+type OpsManager struct {
+	sync.Mutex         // lock to perfom CRUD operations on the cache in a thread safe manner.
+	ready         bool // Ops manager is ready when the cache is ready
+	opsCache      *common.OpsCache
+	cloudProvider gce.GCECompute
+}
+
+func NewOpsManager(c gce.GCECompute) *OpsManager {
+	return &OpsManager{
+		opsCache:      common.NewOpsCache(),
+		cloudProvider: c,
+	}
+}
+
+// ExecuteAttachDisk verifies the cache content before starting an attach disk operation, and then clears the cache if the operation completes successfully.
+func (o *OpsManager) ExecuteAttachDisk(ctx context.Context, opts *AttachDiskOpts) error {
+	// Acquire the ops cache lock, verify ops and proceed to start attach disk operation.
+	op, err := o.checkCacheAndStartAttachDiskOp(ctx, opts.Volumekey, opts.ReadWrite, opts.DiskType, opts.Project, opts.Location, opts.DeviceName, opts.InstanceName)
+	if err != nil {
+		return err
+	}
+
+	// Lock released, wait for the op to complete.
+	err = o.cloudProvider.WaitForZonalOp(ctx, opts.Project, op.Name, opts.Location)
+	if err != nil {
+		return status.Error(codes.Internal, fmt.Sprintf("%v", err))
+	}
+
+	// Op succeeded, acquire cache lock and clear op.
+	o.clearDiskInstanceOpSafe(opts.Project, opts.Location, opts.DeviceName, opts.InstanceName, op.Name)
+	return nil
+}
+
+// ExecuteDetachDisk verifies the cache content before starting a detach disk operation, and then clears the cache if the operation completes successfully.
+func (o *OpsManager) ExecuteDetachDisk(ctx context.Context, opts *DetachDiskOpts) error {
+	// Acquire the ops cache lock, verify ops and proceed to start attach disk operation.
+	op, err := o.checkCacheAndStartDetachDiskOp(ctx, opts.Project, opts.Location, opts.DeviceName, opts.InstanceName)
+	if err != nil {
+		return err
+	}
+
+	// Lock released, wait for the op to complete.
+	err = o.cloudProvider.WaitForZonalOp(ctx, opts.Project, op.Name, opts.Location)
+	if err != nil {
+		return status.Error(codes.Internal, fmt.Sprintf("%v", err))
+	}
+
+	// Op succeeded, acquire cache lock and clear op.
+	o.clearDiskInstanceOpSafe(opts.Project, opts.Location, opts.DeviceName, opts.InstanceName, op.Name)
+	return nil
+}
+
+func (o *OpsManager) checkAndUpdateLastKnownInstanceOps(ctx context.Context, project, location, instanceName string) error {
+	ops := o.opsCache.InstanceOps.GetOps(common.CreateInstanceKey(project, location, instanceName))
+	if ops == nil {
+		return nil
+	}
+
+	for _, op := range ops {
+		klog.V(5).Infof("Found last known op name %q (type %q) for instance %q", op.Name, op.Type, instanceName)
+		done, err := o.cloudProvider.CheckZonalOpDoneStatus(ctx, project, location, op.Name)
+		if err != nil {
+			return status.Error(codes.Internal, fmt.Sprintf("%v", err))
+		}
+
+		if !done {
+			return status.Error(codes.Aborted, fmt.Sprintf("operation %q (type %q) is still in progress", op.Name, op.Type))
+		}
+
+		o.opsCache.InstanceOps.ClearOp(common.CreateInstanceKey(project, location, instanceName), op.Name)
+	}
+
+	return nil
+}
+
+func (o *OpsManager) checkAndUpdateLastKnownDiskInstanceOp(ctx context.Context, project, location, deviceName, instanceName string) error {
+	opInfo := o.opsCache.DiskInstanceOps.GetOp(common.CreateDiskInstanceKey(project, location, deviceName, instanceName))
+	if opInfo == nil {
+		return nil
+	}
+
+	klog.V(5).Infof("Found last known op name %q (type %q) for instance %q", opInfo.Name, opInfo.Type, instanceName)
+	done, err := o.cloudProvider.CheckZonalOpDoneStatus(ctx, project, location, opInfo.Name)
+	if err != nil {
+		return status.Error(codes.Internal, fmt.Sprintf("%v", err))
+	}
+
+	if !done {
+		return status.Error(codes.Aborted, fmt.Sprintf("operation %q (type %q) is still in progress", opInfo.Name, opInfo.Type))
+	}
+
+	o.opsCache.DiskInstanceOps.ClearOp(common.CreateDiskInstanceKey(project, location, deviceName, instanceName), opInfo.Name)
+	return nil
+}
+
+func (o *OpsManager) checkCacheAndStartAttachDiskOp(ctx context.Context, volKey *meta.Key, readWrite, diskType, project, instanceZone, deviceName, instanceName string) (*compute.Operation, error) {
+	o.Lock()
+	defer o.Unlock()
+
+	err := o.checkAndUpdateLastKnownDiskInstanceOp(ctx, project, instanceZone, deviceName, instanceName)
+	if err != nil {
+		return nil, err
+	}
+
+	err = o.checkAndUpdateLastKnownInstanceOps(ctx, project, instanceZone, instanceName)
+	if err != nil {
+		return nil, err
+	}
+
+	op, err := o.cloudProvider.StartAttachDiskOp(ctx, volKey, readWrite, diskType, project, instanceZone, instanceName)
+	if err != nil {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("%v", err))
+	}
+
+	klog.V(5).Infof("AttachDisk operation %q for disk %q started on instance %q", op.Name, deviceName, instanceName)
+	o.opsCache.DiskInstanceOps.AddOp(common.CreateDiskInstanceKey(project, instanceZone, deviceName, instanceName), common.OpInfo{Name: op.Name, Type: op.OperationType})
+	return op, nil
+}
+
+// CheckCacheAndStartDetachDiskOp first verifies that there is no ongoing attach/detach operation for the given disk + instance combination, before triggering a new detach operation.
+func (o *OpsManager) checkCacheAndStartDetachDiskOp(ctx context.Context, project, instanceZone, deviceName, instanceName string) (*compute.Operation, error) {
+	o.Lock()
+	defer o.Unlock()
+
+	err := o.checkAndUpdateLastKnownDiskInstanceOp(ctx, project, instanceZone, deviceName, instanceName)
+	if err != nil {
+		return nil, err
+	}
+
+	err = o.checkAndUpdateLastKnownInstanceOps(ctx, project, instanceZone, instanceName)
+	if err != nil {
+		return nil, err
+	}
+
+	op, err := o.cloudProvider.StartDetachDiskOp(ctx, project, instanceZone, deviceName, instanceName)
+	if err != nil {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("%v", err))
+	}
+
+	klog.V(5).Infof("DetachDisk operation %q for disk %q started on instance %q", op.Name, deviceName, instanceName)
+	o.opsCache.DiskInstanceOps.AddOp(common.CreateDiskInstanceKey(project, instanceZone, deviceName, instanceName), common.OpInfo{Name: op.Name, Type: op.OperationType})
+	return op, nil
+}
+func (o *OpsManager) HydrateOpsCache() {
+	var ops []*compute.Operation
+	retry.OnError(backoff, func(err error) bool { return true }, func() error {
+		var err error
+		ops, err = o.cloudProvider.ListZonalOps(context.Background(), map[string]bool{
+			"attachDisk": true,
+			"detachDisk": true})
+		return err
+	})
+
+	klog.V(5).Infof("Found %d zonal ops", len(ops))
+	o.Lock()
+	defer o.Unlock()
+
+	for _, op := range ops {
+		if done, _ := o.cloudProvider.OpIsDone(op); done {
+			continue
+		}
+		project, zone, instanceName, err := common.ParseOpTargetLinkUrl(op.TargetLink)
+		if err != nil {
+			klog.Errorf("Failed to parse operation target link, err: %s", err)
+			continue
+		}
+
+		klog.V(5).Infof("Adding op %q (type %q) for project %q, zone %q, instance %q to cache", op.Name, op.OperationType, project, zone, instanceName)
+		o.opsCache.InstanceOps.AddOp(common.CreateInstanceKey(project, zone, instanceName), common.OpInfo{Name: op.Name, Type: op.OperationType})
+	}
+
+	o.ready = true
+}
+
+func (o *OpsManager) clearDiskInstanceOpSafe(project, location, diskName, instanceName, opName string) error {
+	o.Lock()
+	defer o.Unlock()
+	return o.opsCache.DiskInstanceOps.ClearOp(common.CreateDiskInstanceKey(project, location, diskName, instanceName), opName)
+}
+
+func (o *OpsManager) IsReady() bool {
+	return o.ready
+}
diff --git a/vendor/k8s.io/client-go/util/retry/OWNERS b/vendor/k8s.io/client-go/util/retry/OWNERS
new file mode 100644
index 00000000..dec3e88d
--- /dev/null
+++ b/vendor/k8s.io/client-go/util/retry/OWNERS
@@ -0,0 +1,4 @@
+# See the OWNERS docs at https://go.k8s.io/owners
+
+reviewers:
+- caesarxuchao
diff --git a/vendor/k8s.io/client-go/util/retry/util.go b/vendor/k8s.io/client-go/util/retry/util.go
new file mode 100644
index 00000000..15e2722f
--- /dev/null
+++ b/vendor/k8s.io/client-go/util/retry/util.go
@@ -0,0 +1,105 @@
+/*
+Copyright 2016 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package retry
+
+import (
+	"time"
+
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/util/wait"
+)
+
+// DefaultRetry is the recommended retry for a conflict where multiple clients
+// are making changes to the same resource.
+var DefaultRetry = wait.Backoff{
+	Steps:    5,
+	Duration: 10 * time.Millisecond,
+	Factor:   1.0,
+	Jitter:   0.1,
+}
+
+// DefaultBackoff is the recommended backoff for a conflict where a client
+// may be attempting to make an unrelated modification to a resource under
+// active management by one or more controllers.
+var DefaultBackoff = wait.Backoff{
+	Steps:    4,
+	Duration: 10 * time.Millisecond,
+	Factor:   5.0,
+	Jitter:   0.1,
+}
+
+// OnError allows the caller to retry fn in case the error returned by fn is retriable
+// according to the provided function. backoff defines the maximum retries and the wait
+// interval between two retries.
+func OnError(backoff wait.Backoff, retriable func(error) bool, fn func() error) error {
+	var lastErr error
+	err := wait.ExponentialBackoff(backoff, func() (bool, error) {
+		err := fn()
+		switch {
+		case err == nil:
+			return true, nil
+		case retriable(err):
+			lastErr = err
+			return false, nil
+		default:
+			return false, err
+		}
+	})
+	if err == wait.ErrWaitTimeout {
+		err = lastErr
+	}
+	return err
+}
+
+// RetryOnConflict is used to make an update to a resource when you have to worry about
+// conflicts caused by other code making unrelated updates to the resource at the same
+// time. fn should fetch the resource to be modified, make appropriate changes to it, try
+// to update it, and return (unmodified) the error from the update function. On a
+// successful update, RetryOnConflict will return nil. If the update function returns a
+// "Conflict" error, RetryOnConflict will wait some amount of time as described by
+// backoff, and then try again. On a non-"Conflict" error, or if it retries too many times
+// and gives up, RetryOnConflict will return an error to the caller.
+//
+//     err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+//         // Fetch the resource here; you need to refetch it on every try, since
+//         // if you got a conflict on the last update attempt then you need to get
+//         // the current version before making your own changes.
+//         pod, err := c.Pods("mynamespace").Get(name, metav1.GetOptions{})
+//         if err ! nil {
+//             return err
+//         }
+//
+//         // Make whatever updates to the resource are needed
+//         pod.Status.Phase = v1.PodFailed
+//
+//         // Try to update
+//         _, err = c.Pods("mynamespace").UpdateStatus(pod)
+//         // You have to return err itself here (not wrapped inside another error)
+//         // so that RetryOnConflict can identify it correctly.
+//         return err
+//     })
+//     if err != nil {
+//         // May be conflict if max retries were hit, or may be something unrelated
+//         // like permissions or a network error
+//         return err
+//     }
+//     ...
+//
+// TODO: Make Backoff an interface?
+func RetryOnConflict(backoff wait.Backoff, fn func() error) error {
+	return OnError(backoff, errors.IsConflict, fn)
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index c35f4730..dbba4200 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -612,6 +612,7 @@ k8s.io/client-go/util/flowcontrol
 k8s.io/client-go/util/homedir
 k8s.io/client-go/util/jsonpath
 k8s.io/client-go/util/keyutil
+k8s.io/client-go/util/retry
 k8s.io/client-go/util/workqueue
 # k8s.io/component-base v0.22.0 => k8s.io/component-base v0.22.0
 ## explicit; go 1.16


Test patch:
diff --git a/pkg/common/cache_test.go b/pkg/common/cache_test.go
new file mode 100644
index 00000000..86704f67
--- /dev/null
+++ b/pkg/common/cache_test.go
@@ -0,0 +1,314 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package common
+
+import (
+	"testing"
+)
+
+type InstanceInfo struct {
+	Project      string
+	Location     string
+	InstanceName string
+}
+
+const (
+	testproject      = "myproject"
+	testinstance     = "myinstance"
+	testdisk         = "mydisk"
+	testlocation     = "us-central1-c"
+	testop           = "op1"
+	attachDiskOpType = "attachDisk"
+	detachDiskOpType = "detachDisk"
+)
+
+func TestKey(t *testing.T) {
+	expectedDiskInstanceKey := DiskInstanceKey(testproject + "_" + testlocation + "_" + testdisk + "_" + testinstance)
+	if CreateDiskInstanceKey(testproject, testlocation, testdisk, testinstance) != expectedDiskInstanceKey {
+		t.Errorf("mismatch in key")
+	}
+
+	expectedInstanceKey := InstanceKey(testproject + "_" + testlocation + "_" + testinstance)
+	if CreateInstanceKey(testproject, testlocation, testinstance) != expectedInstanceKey {
+		t.Errorf("mismatch in key")
+	}
+}
+
+func TestInstanceOpMap(t *testing.T) {
+	dummyInstancekey := InstanceKey("dummykey")
+	// Add operation tests
+	tests := []struct {
+		name        string
+		inputOps    []OpInfo
+		expectedOps map[string]bool
+	}{
+		{
+			name: "add single op",
+			inputOps: []OpInfo{
+				{Name: testop, Type: attachDiskOpType},
+			},
+			expectedOps: map[string]bool{
+				testop: true,
+			},
+		},
+		{
+			name: "add unique ops",
+			inputOps: []OpInfo{
+				{Name: "op-1", Type: attachDiskOpType},
+				{Name: "op-2", Type: attachDiskOpType},
+				{Name: "op-3", Type: attachDiskOpType},
+			},
+			expectedOps: map[string]bool{
+				"op-1": true,
+				"op-2": true,
+				"op-3": true,
+			},
+		},
+		{
+			name: "add duplicate ops, items should be deduped in map",
+			inputOps: []OpInfo{
+				{Name: "op-1", Type: attachDiskOpType},
+				{Name: "op-1", Type: attachDiskOpType},
+				{Name: "op-2", Type: attachDiskOpType},
+				{Name: "op-2", Type: attachDiskOpType},
+			},
+			expectedOps: map[string]bool{
+				"op-1": true,
+				"op-2": true,
+			},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			c := NewOpsCache()
+
+			for _, op := range tc.inputOps {
+				c.InstanceOps.AddOp(dummyInstancekey, op)
+			}
+
+			if len(c.InstanceOps.GetOps(dummyInstancekey)) != len(tc.expectedOps) {
+				t.Errorf("mismatch in expected number of ops in map")
+			}
+
+			for _, op := range c.InstanceOps.GetOps(dummyInstancekey) {
+				if _, ok := tc.expectedOps[op.Name]; !ok {
+					t.Errorf("unexpected op %q", op.Name)
+				}
+			}
+		})
+	}
+
+	// Clear operation tests
+	tests1 := []struct {
+		name            string
+		inputOps        []OpInfo
+		clearOps        []string
+		expectedOps     map[string]bool
+		expectKeyRemove bool
+	}{
+		{
+			name:        "empty input map, clear item is a no-op",
+			inputOps:    []OpInfo{},
+			clearOps:    []string{"op-1", "op-2"},
+			expectedOps: map[string]bool{},
+		},
+		{
+			name: "non-empty input map, clear item for non-existent key is no-op",
+			inputOps: []OpInfo{
+				{Name: "op-1", Type: attachDiskOpType},
+				{Name: "op-2", Type: attachDiskOpType},
+			},
+			clearOps: []string{"op-3"},
+			expectedOps: map[string]bool{
+				"op-1": true,
+				"op-2": true,
+			},
+		},
+		{
+			name: "non-empty input map, clear item for existent key, 0 remaining values",
+			inputOps: []OpInfo{
+				{Name: "op-1", Type: attachDiskOpType},
+				{Name: "op-2", Type: attachDiskOpType},
+			},
+			clearOps:        []string{"op-1", "op-2"},
+			expectedOps:     map[string]bool{},
+			expectKeyRemove: true,
+		},
+		{
+			name: "non-empty input map, clear item for existent key, non 0 remaining values",
+			inputOps: []OpInfo{
+				{Name: "op-1", Type: attachDiskOpType},
+				{Name: "op-2", Type: attachDiskOpType},
+			},
+			clearOps: []string{"op-1"},
+			expectedOps: map[string]bool{
+				"op-2": true,
+			},
+		},
+	}
+	for _, tc := range tests1 {
+		t.Run(tc.name, func(t *testing.T) {
+			c := NewOpsCache()
+			for _, op := range tc.inputOps {
+				c.InstanceOps.AddOp(dummyInstancekey, op)
+			}
+
+			for _, op := range tc.clearOps {
+				c.InstanceOps.ClearOp(dummyInstancekey, op)
+			}
+
+			if tc.expectKeyRemove {
+				_, ok := c.InstanceOps.ops[dummyInstancekey]
+				if ok {
+					t.Errorf("unexpected key found")
+				}
+			}
+
+			if len(c.InstanceOps.GetOps(dummyInstancekey)) != len(tc.expectedOps) {
+				t.Errorf("mismatch in expected number of ops in map")
+			}
+			for _, op := range c.InstanceOps.GetOps(dummyInstancekey) {
+				if _, ok := tc.expectedOps[op.Name]; !ok {
+					t.Errorf("unexpected op %q", op.Name)
+				}
+			}
+		})
+	}
+}
+
+func TestDiskInstanceOpMap(t *testing.T) {
+	// Add operation tests
+	type CacheEntry struct {
+		Key DiskInstanceKey
+		Op  OpInfo
+	}
+	tests := []struct {
+		name             string
+		inputEntries     []CacheEntry
+		clearEntry       CacheEntry
+		expectedEntries  map[DiskInstanceKey]OpInfo
+		expectClearOpErr bool
+	}{
+		{
+			name: "clear called on empty map",
+			clearEntry: CacheEntry{
+				Key: "key1",
+				Op: OpInfo{
+					Name: "op-1",
+					Type: attachDiskOpType,
+				},
+			},
+		},
+		{
+			name: "clear called on non-empty map",
+			inputEntries: []CacheEntry{
+				{
+					Key: "key1",
+					Op: OpInfo{
+						Name: "op-1",
+						Type: attachDiskOpType,
+					},
+				},
+				{
+					Key: "key2",
+					Op: OpInfo{
+						Name: "op-2",
+						Type: attachDiskOpType,
+					},
+				},
+			},
+			clearEntry: CacheEntry{
+				Key: "key1",
+				Op: OpInfo{
+					Name: "op-1",
+				},
+			},
+			expectedEntries: map[DiskInstanceKey]OpInfo{
+				"key2": {
+					Name: "op-2",
+					Type: attachDiskOpType,
+				},
+			},
+		},
+		{
+			name: "clear called on a key with different op name, error expected",
+			inputEntries: []CacheEntry{
+				{
+					Key: "key1",
+					Op: OpInfo{
+						Name: "op-1",
+						Type: attachDiskOpType,
+					},
+				},
+				{
+					Key: "key2",
+					Op: OpInfo{
+						Name: "op-2",
+						Type: attachDiskOpType,
+					},
+				},
+			},
+			clearEntry: CacheEntry{
+				Key: "key1",
+				Op: OpInfo{
+					Name: "op-3",
+				},
+			},
+			expectedEntries: map[DiskInstanceKey]OpInfo{
+				"key1": {
+					Name: "op-1",
+					Type: attachDiskOpType,
+				},
+				"key2": {
+					Name: "op-2",
+					Type: attachDiskOpType,
+				},
+			},
+			expectClearOpErr: true,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			c := NewOpsCache()
+			for _, e := range tc.inputEntries {
+				c.DiskInstanceOps.AddOp(e.Key, e.Op)
+			}
+
+			err := c.DiskInstanceOps.ClearOp(tc.clearEntry.Key, tc.clearEntry.Op.Name)
+			if tc.expectClearOpErr && err == nil {
+				t.Errorf("Expected error not found")
+			}
+			if !tc.expectClearOpErr && err != nil {
+				t.Errorf("Unexpected error found")
+			}
+
+			if len(tc.expectedEntries) != len(c.DiskInstanceOps.ops) {
+				t.Errorf("Unexpected number of keys found")
+			}
+
+			for k, v := range c.DiskInstanceOps.ops {
+				expectedV, ok := tc.expectedEntries[k]
+				if !ok {
+					t.Errorf("unexpected key %q", k)
+				}
+				if v.Name != expectedV.Name || v.Type != expectedV.Type {
+					t.Errorf("unexpected value %q, %q", v.Name, v.Type)
+				}
+			}
+		})
+	}
+}
diff --git a/pkg/common/utils_test.go b/pkg/common/utils_test.go
index ca435caf..181b8c72 100644
--- a/pkg/common/utils_test.go
+++ b/pkg/common/utils_test.go
@@ -577,3 +577,78 @@ func TestSnapshotStorageLocations(t *testing.T) {
 		})
 	}
 }
+
+func TestParseOpTargetLinkUrl(t *testing.T) {
+	type InstanceInfo struct {
+		project  string
+		location string
+		name     string
+	}
+	tests := []struct {
+		name             string
+		url              string
+		expectError      bool
+		expectedInstance *InstanceInfo
+	}{
+		{
+			name:        "empty url",
+			url:         "",
+			expectError: true,
+		},
+		{
+			name:        "invalid instance 1",
+			url:         "https://www.googleapis.co/compute/v1beta1/projects/myproject/zones/us-central1-c/instances/myinstance",
+			expectError: true,
+		},
+		{
+			name:        "invalid instance 2",
+			url:         "https://www.googleapis.comcompute/v1beta1/projects/myproject/zones/us-central1-c/instances/myinstance",
+			expectError: true,
+		},
+		{
+			name:        "invalid instance 3",
+			url:         "https://www.googleapis.com/compute/v1beta1/myproject/zones/us-central1-c/instances/myinstance",
+			expectError: true,
+		},
+		{
+			name:        "invalid instance 4",
+			url:         "https://www.googleapis.com/compute/v1beta1/projects/myproject/us-central1-c/instances/myinstance",
+			expectError: true,
+		},
+		{
+			name: "valid v1 instance",
+			url:  "https://www.googleapis.com/compute/v1/projects/myproject/zones/us-central1-c/instances/myinstance",
+			expectedInstance: &InstanceInfo{
+				project:  "myproject",
+				name:     "myinstance",
+				location: "us-central1-c",
+			},
+		},
+		{
+			name: "valid v1beta1 instance",
+			url:  "https://www.googleapis.com/compute/v1beta1/projects/myproject/zones/us-central1-c/instances/myinstance",
+			expectedInstance: &InstanceInfo{
+				project:  "myproject",
+				name:     "myinstance",
+				location: "us-central1-c",
+			},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			project, location, instance, err := ParseOpTargetLinkUrl(tc.url)
+			if tc.expectError && err == nil {
+				t.Error("expected error")
+			}
+			if !tc.expectError && err != nil {
+				t.Error("unexpected error")
+			}
+
+			if tc.expectedInstance != nil {
+				if tc.expectedInstance.project != project || tc.expectedInstance.name != instance || tc.expectedInstance.location != location {
+					t.Error("unexpected instance found")
+				}
+			}
+		})
+	}
+}
diff --git a/pkg/gce-pd-csi-driver/controller_test.go b/pkg/gce-pd-csi-driver/controller_test.go
index 14f20361..7322719d 100644
--- a/pkg/gce-pd-csi-driver/controller_test.go
+++ b/pkg/gce-pd-csi-driver/controller_test.go
@@ -32,10 +32,12 @@ import (
 	"google.golang.org/grpc/status"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/util/retry"
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	"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"
+	gcecloudprovider "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute"
 )
 
 const (
@@ -1693,7 +1695,7 @@ func TestPickRandAndConsecutive(t *testing.T) {
 }
 
 func TestVolumeOperationConcurrency(t *testing.T) {
-	readyToExecute := make(chan chan struct{}, 1)
+	readyToExecute := make(chan chan gcecloudprovider.Signal, 1)
 	gceDriver := initBlockingGCEDriver(t, []*gce.CloudDisk{
 		createZonalCloudDisk(name + "1"),
 		createZonalCloudDisk(name + "2"),
@@ -1750,13 +1752,13 @@ func TestVolumeOperationConcurrency(t *testing.T) {
 	// Start vol2CreateSnapshot and allow it to execute to completion. Then check for success.
 	vol2CreateSnapshotResp := runRequest(vol2CreateSnapshotReq)
 	execVol2CreateSnapshot := <-readyToExecute
-	execVol2CreateSnapshot <- struct{}{}
+	execVol2CreateSnapshot <- gcecloudprovider.Signal{}
 	if err := <-vol2CreateSnapshotResp; err != nil {
 		t.Errorf("Unexpected error: %v", err)
 	}
 
 	// To clean up, allow the vol1CreateSnapshotA to complete
-	execVol1CreateSnapshotA <- struct{}{}
+	execVol1CreateSnapshotA <- gcecloudprovider.Signal{}
 	if err := <-vol1CreateSnapshotAResp; err != nil {
 		t.Errorf("Unexpected error: %v", err)
 	}
@@ -1998,7 +2000,7 @@ func TestControllerPublishUnpublishVolume(t *testing.T) {
 		} else {
 			gceDriver = initGCEDriver(t, tc.seedDisks)
 		}
-
+		gceDriver.cs.opsManager.ready = true
 		// mark the node in the map
 		if tc.errorSeenOnNode {
 			gceDriver.cs.publishErrorsSeenOnNode[testNodeID] = true
@@ -2051,3 +2053,1062 @@ func TestControllerPublishUnpublishVolume(t *testing.T) {
 		}
 	}
 }
+
+func TestControllerPublishInterop(t *testing.T) {
+	readyToExecute := make(chan chan gcecloudprovider.Signal, 1)
+	disk1 := name + "1"
+	disk2 := name + "2"
+	cloudDisks := []*gce.CloudDisk{
+		createZonalCloudDisk(disk1),
+		createZonalCloudDisk(disk2),
+	}
+	fcp, err := gce.CreateFakeCloudProvider(project, zone, cloudDisks)
+	if err != nil {
+		t.Fatalf("Failed to create fake cloud provider: %v", err)
+	}
+	fcpBlocking := &gce.FakeBlockingCloudProvider{
+		FakeCloudProvider: fcp,
+		ReadyToExecute:    readyToExecute,
+	}
+	instance := &compute.Instance{
+		Name:  node,
+		Disks: []*compute.AttachedDisk{},
+	}
+	fcp.InsertInstance(instance, zone, node)
+	gceDriver := initGCEDriverWithCloudProvider(t, fcpBlocking)
+	cs := gceDriver.cs
+	cs.opsManager.ready = true
+	runRequest := func(req *csi.ControllerPublishVolumeRequest) <-chan error {
+		response := make(chan error)
+		go func() {
+			_, err := cs.ControllerPublishVolume(context.Background(), req)
+			response <- err
+		}()
+		return response
+	}
+
+	vol1node1PublishReq := &csi.ControllerPublishVolumeRequest{
+		VolumeId: testVolumeID + "1",
+		NodeId:   testNodeID,
+		VolumeCapability: &csi.VolumeCapability{
+			AccessType: &csi.VolumeCapability_Mount{
+				Mount: &csi.VolumeCapability_MountVolume{},
+			},
+			AccessMode: &csi.VolumeCapability_AccessMode{
+				Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+			},
+		},
+	}
+
+	// Run controller publish request. This is expected to do following:
+	// 1. check cache, find no current running ops for disk+instance key
+	// 2. Start AttachDisk, update cache with op details
+	// 3. release cache lock
+	// 4. block on the mock Poll.
+	vol1node1ControllerPublishResp := runRequest(vol1node1PublishReq)
+	executeVol1ControllerPublish := <-readyToExecute
+
+	//  Verify cache content
+	diskInstanceKey := common.CreateDiskInstanceKey(project, zone, disk1, node)
+	op := cs.opsManager.opsCache.DiskInstanceOps.GetOp(diskInstanceKey)
+	if op == nil {
+		t.Errorf("expected attach op in cache")
+	}
+	if op.Type != "attachDisk" {
+		t.Errorf("Unexpected op %s, %s found", op.Name, op.Type)
+	}
+
+	// Start controller publish request for vol2 on node1. This is expected to do the following:
+	// 1. Check cache and find no entry for disk+instance
+	// 2. Start AttachDisk, update cache with op details
+	// 3. release cache lock
+	// 4. block on the mock Poll.
+	vol2node1PublishReq := &csi.ControllerPublishVolumeRequest{
+		VolumeId: testVolumeID + "2",
+		NodeId:   testNodeID,
+		VolumeCapability: &csi.VolumeCapability{
+			AccessType: &csi.VolumeCapability_Mount{
+				Mount: &csi.VolumeCapability_MountVolume{},
+			},
+			AccessMode: &csi.VolumeCapability_AccessMode{
+				Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+			},
+		},
+	}
+
+	vol2node1ControllerPublishResp := runRequest(vol2node1PublishReq)
+	executeVol2ControllerPublish := <-readyToExecute
+	disk1InstanceKey := common.CreateDiskInstanceKey(project, zone, disk1, node)
+	disk2InstanceKey := common.CreateDiskInstanceKey(project, zone, disk2, node)
+	op1 := cs.opsManager.opsCache.DiskInstanceOps.GetOp(disk1InstanceKey)
+	if op1 == nil {
+		t.Errorf("expected attach op in cache for disk %q", disk1)
+	}
+	if op1.Type != "attachDisk" {
+		t.Errorf("Unexpected op %s, %s found", op.Name, op.Type)
+	}
+	op2 := cs.opsManager.opsCache.DiskInstanceOps.GetOp(disk2InstanceKey)
+	if op2 == nil {
+		t.Errorf("expected attach op in cache")
+	}
+	if op2.Type != "attachDisk" {
+		t.Errorf("Unexpected op %s, %s found", op.Name, op.Type)
+	}
+
+	// unblock execute of controller publish of first volume
+	s := gcecloudprovider.Signal{}
+	executeVol1ControllerPublish <- s
+	if err := <-vol1node1ControllerPublishResp; err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+
+	// verify cache
+	op1 = cs.opsManager.opsCache.DiskInstanceOps.GetOp(disk1InstanceKey)
+	if op1 != nil {
+		t.Errorf("unexpected attach op in cache")
+	}
+
+	// unblock execute of controller publish of second volume. Mock error in poll. The CSI op will return failure and mark the node with failures.
+	s1 := gcecloudprovider.Signal{ReportError: true}
+	executeVol2ControllerPublish <- s1
+	if err := <-vol2node1ControllerPublishResp; err == nil {
+		t.Errorf("expected error")
+	}
+
+	// verify cache still contains op for vol2.
+	op2 = cs.opsManager.opsCache.DiskInstanceOps.GetOp(disk2InstanceKey)
+	if op2 == nil {
+		t.Errorf("Expected attach op in cache for disk %q", disk2)
+	}
+	if op2.Type != "attachDisk" {
+		t.Errorf("Unexpected op %s, %s found", op.Name, op.Type)
+	}
+}
+
+type DiskInstanceOpCacheEntry struct {
+	Key common.DiskInstanceKey
+	Op  common.OpInfo
+}
+type InstanceOpCacheEntry struct {
+	Key common.InstanceKey
+	Op  common.OpInfo
+}
+
+func TestControllerPublishUnpublishDiskInstanceOpCache(t *testing.T) {
+	disk1 := name + "1"
+	volId1 := testVolumeID + "1"
+	containsDiskInstanceEntry := func(cache *common.OpsCache, e DiskInstanceOpCacheEntry) bool {
+		op := cache.DiskInstanceOps.GetOp(e.Key)
+		if op == nil {
+			return false
+		}
+
+		if op.Name != e.Op.Name || op.Type != e.Op.Type {
+			return false
+		}
+		return true
+	}
+
+	tests := []struct {
+		name                        string
+		initialCache                []DiskInstanceOpCacheEntry // pre-populate cache before calling CSI op.
+		expectAttachDetachOpInCache bool                       // While poll filestore op is in progress, we expect the op to be present in cache.
+		pubReq                      *csi.ControllerPublishVolumeRequest
+		unpubReq                    *csi.ControllerUnpublishVolumeRequest
+		checkOpStatusError          bool // Whether to return an error during check of op status
+		checkOpStatusRunning        bool // Whether to return a running status during check of op status
+		signalCheckOpDone           bool // if the code path is expected to reach the check op status, the fake blocking cloud provider will block for a signal to proceed forward.
+		signalPollOp                bool // if the code path is expected to reach the stage where the filestore op is polled, the fake blocking cloud provider will block for a signal to proceed forward.
+		pollOpDoneError             bool // whether to return an error to mock poll error.
+		expectCSIOpError            bool // whether the high level csi op is expected to fail
+	}{
+		{
+			name: "empty cache, publish op completes successfully",
+			pubReq: &csi.ControllerPublishVolumeRequest{
+				VolumeId: volId1,
+				NodeId:   testNodeID,
+				VolumeCapability: &csi.VolumeCapability{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+					},
+				},
+			},
+			signalPollOp: true,
+		},
+		{
+			name: "non-empty initial cache, check op status returns error for the attach op, csi publish op returns error",
+			initialCache: []DiskInstanceOpCacheEntry{
+				{
+					Key: common.CreateDiskInstanceKey(project, zone, disk1, node),
+					Op: common.OpInfo{
+						Name: "op-1",
+						Type: "attachDisk",
+					},
+				},
+			},
+			pubReq: &csi.ControllerPublishVolumeRequest{
+				VolumeId: volId1,
+				NodeId:   testNodeID,
+				VolumeCapability: &csi.VolumeCapability{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+					},
+				},
+			},
+			signalCheckOpDone:  true,
+			checkOpStatusError: true,
+			expectCSIOpError:   true,
+		},
+		{
+			name: "non-empty initial cache, attach op in progress, csi publish op returns error",
+			initialCache: []DiskInstanceOpCacheEntry{
+				{
+					Key: common.CreateDiskInstanceKey(project, zone, disk1, node),
+					Op: common.OpInfo{
+						Name: "op-1",
+						Type: "attachDisk",
+					},
+				},
+			},
+			pubReq: &csi.ControllerPublishVolumeRequest{
+				VolumeId: volId1,
+				NodeId:   testNodeID,
+				VolumeCapability: &csi.VolumeCapability{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+					},
+				},
+			},
+			signalCheckOpDone:    true,
+			checkOpStatusRunning: true,
+			expectCSIOpError:     true,
+		},
+		{
+			name: "non-empty initial cache, detach op check returns error, csi publish op returns error",
+			initialCache: []DiskInstanceOpCacheEntry{
+				{
+					Key: common.CreateDiskInstanceKey(project, zone, disk1, node),
+					Op: common.OpInfo{
+						Name: "op-1",
+						Type: "detachDisk",
+					},
+				},
+			},
+			pubReq: &csi.ControllerPublishVolumeRequest{
+				VolumeId: volId1,
+				NodeId:   testNodeID,
+				VolumeCapability: &csi.VolumeCapability{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+					},
+				},
+			},
+			signalCheckOpDone:  true,
+			checkOpStatusError: true,
+			expectCSIOpError:   true,
+		},
+		{
+			name: "non-empty initial cache, detach op in progress for same disk+instance, csi publish op returns error",
+			initialCache: []DiskInstanceOpCacheEntry{
+				{
+					Key: common.CreateDiskInstanceKey(project, zone, disk1, node),
+					Op: common.OpInfo{
+						Name: "op-1",
+						Type: "detachDisk",
+					},
+				},
+			},
+			pubReq: &csi.ControllerPublishVolumeRequest{
+				VolumeId: volId1,
+				NodeId:   testNodeID,
+				VolumeCapability: &csi.VolumeCapability{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+					},
+				},
+			},
+			signalCheckOpDone:    true,
+			checkOpStatusRunning: true,
+			expectCSIOpError:     true,
+		},
+		{
+			name: "non-empty initial cache, detach op complete for same disk+instance, csi publish op returns success",
+			initialCache: []DiskInstanceOpCacheEntry{
+				{
+					Key: common.CreateDiskInstanceKey(project, zone, disk1, node),
+					Op: common.OpInfo{
+						Name: "op-1",
+						Type: "detachDisk",
+					},
+				},
+			},
+			pubReq: &csi.ControllerPublishVolumeRequest{
+				VolumeId: volId1,
+				NodeId:   testNodeID,
+				VolumeCapability: &csi.VolumeCapability{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+					},
+				},
+			},
+			signalCheckOpDone:           true,
+			signalPollOp:                true,
+			expectAttachDetachOpInCache: true,
+		},
+		{
+			name: "non-empty initial cache, attach op complete for same disk+instance, csi publish op returns success",
+			initialCache: []DiskInstanceOpCacheEntry{
+				{
+					Key: common.CreateDiskInstanceKey(project, zone, disk1, node),
+					Op: common.OpInfo{
+						Name: "op-1",
+						Type: "attachDisk",
+					},
+				},
+			},
+			pubReq: &csi.ControllerPublishVolumeRequest{
+				VolumeId: volId1,
+				NodeId:   testNodeID,
+				VolumeCapability: &csi.VolumeCapability{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+					},
+				},
+			},
+			signalCheckOpDone:           true,
+			signalPollOp:                true,
+			expectAttachDetachOpInCache: true,
+		},
+		// Unpubish ops
+		{
+			name: "empty cache, unpublish op completes successfully",
+			unpubReq: &csi.ControllerUnpublishVolumeRequest{
+				VolumeId: volId1,
+				NodeId:   testNodeID,
+			},
+			signalPollOp: true,
+		},
+		{
+			name: "non-empty initial cache, detach op check returns error, csi publish op returns error",
+			initialCache: []DiskInstanceOpCacheEntry{
+				{
+					Key: common.CreateDiskInstanceKey(project, zone, disk1, node),
+					Op: common.OpInfo{
+						Name: "op-1",
+						Type: "detachDisk",
+					},
+				},
+			},
+			unpubReq: &csi.ControllerUnpublishVolumeRequest{
+				VolumeId: volId1,
+				NodeId:   testNodeID,
+			},
+			signalCheckOpDone:  true,
+			checkOpStatusError: true,
+			expectCSIOpError:   true,
+		},
+		{
+			name: "non-empty initial cache, detach op in progress for same disk+instance, csi publish op returns error",
+			initialCache: []DiskInstanceOpCacheEntry{
+				{
+					Key: common.CreateDiskInstanceKey(project, zone, disk1, node),
+					Op: common.OpInfo{
+						Name: "op-1",
+						Type: "detachDisk",
+					},
+				},
+			},
+			unpubReq: &csi.ControllerUnpublishVolumeRequest{
+				VolumeId: volId1,
+				NodeId:   testNodeID,
+			},
+			signalCheckOpDone:    true,
+			checkOpStatusRunning: true,
+			expectCSIOpError:     true,
+		},
+		{
+			name: "non-empty initial cache, attach op check returns error, csi publish op returns error",
+			initialCache: []DiskInstanceOpCacheEntry{
+				{
+					Key: common.CreateDiskInstanceKey(project, zone, disk1, node),
+					Op: common.OpInfo{
+						Name: "op-1",
+						Type: "attachDisk",
+					},
+				},
+			},
+			unpubReq: &csi.ControllerUnpublishVolumeRequest{
+				VolumeId: volId1,
+				NodeId:   testNodeID,
+			},
+			signalCheckOpDone:  true,
+			checkOpStatusError: true,
+			expectCSIOpError:   true,
+		},
+		{
+			name: "non-empty initial cache, attach op in progress for same disk+instance, csi publish op returns error",
+			initialCache: []DiskInstanceOpCacheEntry{
+				{
+					Key: common.CreateDiskInstanceKey(project, zone, disk1, node),
+					Op: common.OpInfo{
+						Name: "op-1",
+						Type: "attachDisk",
+					},
+				},
+			},
+			unpubReq: &csi.ControllerUnpublishVolumeRequest{
+				VolumeId: volId1,
+				NodeId:   testNodeID,
+			},
+			signalCheckOpDone:    true,
+			checkOpStatusRunning: true,
+			expectCSIOpError:     true,
+		},
+		{
+			name: "non-empty initial cache, detach op complete for same disk+instance, csi unpublish op returns success",
+			initialCache: []DiskInstanceOpCacheEntry{
+				{
+					Key: common.CreateDiskInstanceKey(project, zone, disk1, node),
+					Op: common.OpInfo{
+						Name: "op-1",
+						Type: "detachDisk",
+					},
+				},
+			},
+			unpubReq: &csi.ControllerUnpublishVolumeRequest{
+				VolumeId: volId1,
+				NodeId:   testNodeID,
+			},
+			signalCheckOpDone:           true,
+			signalPollOp:                true,
+			expectAttachDetachOpInCache: true,
+		},
+		{
+			name: "non-empty initial cache, attach op complete for same disk+instance, csi unpublish op returns success",
+			initialCache: []DiskInstanceOpCacheEntry{
+				{
+					Key: common.CreateDiskInstanceKey(project, zone, disk1, node),
+					Op: common.OpInfo{
+						Name: "op-1",
+						Type: "attachDisk",
+					},
+				},
+			},
+			unpubReq: &csi.ControllerUnpublishVolumeRequest{
+				VolumeId: volId1,
+				NodeId:   testNodeID,
+			},
+			signalCheckOpDone:           true,
+			signalPollOp:                true,
+			expectAttachDetachOpInCache: true,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			readyToExecute := make(chan chan gcecloudprovider.Signal, 1)
+
+			cloudDisks := []*gce.CloudDisk{
+				createZonalCloudDisk(disk1),
+			}
+			fcp, err := gce.CreateFakeCloudProvider(project, zone, cloudDisks)
+			if err != nil {
+				t.Fatalf("Failed to create fake cloud provider: %v", err)
+			}
+			fcpBlocking := &gce.FakeBlockingCloudProvider{
+				FakeCloudProvider: fcp,
+				ReadyToExecute:    readyToExecute,
+			}
+			instance := &compute.Instance{
+				Name:  node,
+				Disks: []*compute.AttachedDisk{},
+			}
+			if tc.unpubReq != nil {
+				instance.Disks = append(instance.Disks, &compute.AttachedDisk{DeviceName: disk1})
+			}
+			fcp.InsertInstance(instance, zone, node)
+			gceDriver := initGCEDriverWithCloudProvider(t, fcpBlocking)
+			cs := gceDriver.cs
+			for _, e := range tc.initialCache {
+				cs.opsManager.opsCache.DiskInstanceOps.AddOp(e.Key, e.Op)
+			}
+			cs.opsManager.ready = true
+
+			runPublishRequest := func(req *csi.ControllerPublishVolumeRequest) <-chan error {
+				response := make(chan error)
+				go func() {
+					_, err := cs.ControllerPublishVolume(context.Background(), req)
+					response <- err
+				}()
+				return response
+			}
+			runUnpublishRequest := func(req *csi.ControllerUnpublishVolumeRequest) <-chan error {
+				response := make(chan error)
+				go func() {
+					_, err := cs.ControllerUnpublishVolume(context.Background(), req)
+					response <- err
+				}()
+				return response
+			}
+
+			var resp <-chan error
+			if tc.pubReq != nil {
+				resp = runPublishRequest(tc.pubReq)
+			} else if tc.unpubReq != nil {
+				resp = runUnpublishRequest(tc.unpubReq)
+			} else {
+				t.Errorf("invalid test case")
+			}
+
+			// If a key corresponding to the disk+instance found in the cache, controller will check the op status.
+			if tc.signalCheckOpDone {
+				s := gcecloudprovider.Signal{}
+				if tc.checkOpStatusError {
+					s.ReportError = true
+				} else if tc.checkOpStatusRunning {
+					s.ReportRunning = true
+				}
+				execute := <-readyToExecute
+				execute <- s
+			}
+
+			if tc.expectAttachDetachOpInCache {
+				// Find the running attach/detach op. This may need a few retries, because, at this time the controller publish/unpublish op will the cache and add a new op to cache.
+				backoff := wait.Backoff{
+					Duration: 10 * time.Millisecond,
+					Steps:    100,
+				}
+				if err := retry.OnError(backoff, func(err error) bool { return true }, func() error {
+					ops, innererr := cs.CloudProvider.ListZonalOps(context.Background(), map[string]bool{
+						"attachDisk": true, "detachDisk": true})
+					if innererr != nil {
+						return innererr
+					}
+					if len(ops) > 0 {
+						return nil
+					}
+					return fmt.Errorf("failed to find ops for fake cloud provider")
+				}); err != nil {
+					t.Errorf("timed out waiting for attach op to be updated")
+					return
+				}
+				ops, err := cs.CloudProvider.ListZonalOps(context.Background(), map[string]bool{"attachDisk": true})
+				if err != nil {
+					t.Errorf("Unexpected error in finding ops")
+				}
+				if len(ops) != 1 {
+					t.Errorf("Unexpected number of attach ops in cache")
+					return
+				}
+				// Now the controller has updated the cache and should block at poll
+				// verify cache content
+				opinfo := common.OpInfo{
+					Name: ops[0].Name,
+					Type: ops[0].OperationType,
+				}
+				if !containsDiskInstanceEntry(cs.opsManager.opsCache, DiskInstanceOpCacheEntry{Key: common.CreateDiskInstanceKey(project, zone, disk1, node), Op: opinfo}) {
+					t.Errorf("Unexpected cache entry detected")
+				}
+			}
+
+			// Unblock the poll operation
+			if tc.signalPollOp {
+				s := gcecloudprovider.Signal{}
+				if tc.pollOpDoneError {
+					s.ReportError = true
+				}
+				execute := <-readyToExecute
+				execute <- s
+			}
+
+			err = <-resp
+			if tc.expectCSIOpError && err == nil {
+				t.Errorf("Expected error found none")
+			}
+			if !tc.expectCSIOpError && err != nil {
+				t.Errorf("Unexpected error found")
+			}
+		})
+	}
+}
+
+func TestControllerPublishUnpublishInstanceOpCache(t *testing.T) {
+	type InstanceCacheEntry struct {
+		Key common.InstanceKey
+		Op  common.OpInfo
+	}
+	disk1 := name + "1"
+	volId1 := testVolumeID + "1"
+	containsDiskInstanceEntry := func(cache *common.OpsCache, e DiskInstanceOpCacheEntry) bool {
+		op := cache.DiskInstanceOps.GetOp(e.Key)
+		if op == nil {
+			return false
+		}
+
+		if op.Name != e.Op.Name || op.Type != e.Op.Type {
+			return false
+		}
+		return true
+	}
+
+	tests := []struct {
+		name                        string
+		initialCache                []InstanceOpCacheEntry // pre-populate cache before calling CSI op.
+		expectAttachDetachOpInCache bool                   // While poll filestore op is in progress, we expect the op to be present in cache.
+		pubReq                      *csi.ControllerPublishVolumeRequest
+		unpubReq                    *csi.ControllerUnpublishVolumeRequest
+		checkOpStatusError          []bool // sequence of error to return during status check for each op.
+		checkOpStatusRunning        []bool // sequence of done status to return during status check for each op.
+		signalCheckOpDoneCount      int    // Expected number of times, the tester would block on check status op.
+		signalPollOp                bool   // if the code path is expected to reach the stage where the filestore op is polled, the fake blocking cloud provider will block for a signal to proceed forward.
+		expectCSIOpError            bool   // whether the high level csi op is expected to fail
+	}{
+		{
+			name: "empty cache, publish op completes successfully",
+			pubReq: &csi.ControllerPublishVolumeRequest{
+				VolumeId: volId1,
+				NodeId:   testNodeID,
+				VolumeCapability: &csi.VolumeCapability{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+					},
+				},
+			},
+			expectAttachDetachOpInCache: true,
+			signalPollOp:                true,
+		},
+		{
+			name: "non-empty initial cache, check op status on same instance returns error, csi publish op returns error",
+			initialCache: []InstanceOpCacheEntry{
+				{
+					Key: common.CreateInstanceKey(project, zone, node),
+					Op: common.OpInfo{
+						Name: "op-1",
+						Type: "attachDisk",
+					},
+				},
+			},
+			pubReq: &csi.ControllerPublishVolumeRequest{
+				VolumeId: volId1,
+				NodeId:   testNodeID,
+				VolumeCapability: &csi.VolumeCapability{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+					},
+				},
+			},
+			signalCheckOpDoneCount: 1,
+			checkOpStatusError:     []bool{true},
+			expectCSIOpError:       true,
+		},
+		{
+			name: "non-empty initial cache, check op status running on same instance, csi publish op returns error",
+			initialCache: []InstanceOpCacheEntry{
+				{
+					Key: common.CreateInstanceKey(project, zone, node),
+					Op: common.OpInfo{
+						Name: "op-1",
+						Type: "attachDisk",
+					},
+				},
+			},
+			pubReq: &csi.ControllerPublishVolumeRequest{
+				VolumeId: volId1,
+				NodeId:   testNodeID,
+				VolumeCapability: &csi.VolumeCapability{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+					},
+				},
+			},
+			signalCheckOpDoneCount: 1,
+			checkOpStatusRunning:   []bool{true},
+			expectCSIOpError:       true,
+		},
+		{
+			name: "non-empty initial cache, subset of ops for same instance returns error, csi publish op returns error",
+			initialCache: []InstanceOpCacheEntry{
+				{
+					Key: common.CreateInstanceKey(project, zone, node),
+					Op: common.OpInfo{
+						Name: "op-1",
+						Type: "attachDisk",
+					},
+				},
+				{
+					Key: common.CreateInstanceKey(project, zone, node),
+					Op: common.OpInfo{
+						Name: "op-2",
+						Type: "detachDisk",
+					},
+				},
+			},
+			pubReq: &csi.ControllerPublishVolumeRequest{
+				VolumeId: volId1,
+				NodeId:   testNodeID,
+				VolumeCapability: &csi.VolumeCapability{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+					},
+				},
+			},
+			checkOpStatusError: []bool{false, true},
+			expectCSIOpError:   true,
+		},
+		{
+			name: "non-empty initial cache, subset of ops for same instance still in progress, csi publish op returns error",
+			initialCache: []InstanceOpCacheEntry{
+				{
+					Key: common.CreateInstanceKey(project, zone, node),
+					Op: common.OpInfo{
+						Name: "op-1",
+						Type: "attachDisk",
+					},
+				},
+				{
+					Key: common.CreateInstanceKey(project, zone, node),
+					Op: common.OpInfo{
+						Name: "op-2",
+						Type: "detachDisk",
+					},
+				},
+			},
+			pubReq: &csi.ControllerPublishVolumeRequest{
+				VolumeId: volId1,
+				NodeId:   testNodeID,
+				VolumeCapability: &csi.VolumeCapability{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+					},
+				},
+			},
+			checkOpStatusRunning: []bool{false, true},
+			expectCSIOpError:     true,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			readyToExecute := make(chan chan gcecloudprovider.Signal, 1)
+
+			cloudDisks := []*gce.CloudDisk{
+				createZonalCloudDisk(disk1),
+			}
+			fcp, err := gce.CreateFakeCloudProvider(project, zone, cloudDisks)
+			if err != nil {
+				t.Fatalf("Failed to create fake cloud provider: %v", err)
+			}
+			fcpBlocking := &gce.FakeBlockingCloudProvider{
+				FakeCloudProvider: fcp,
+				ReadyToExecute:    readyToExecute,
+			}
+			instance := &compute.Instance{
+				Name:  node,
+				Disks: []*compute.AttachedDisk{},
+			}
+			if tc.unpubReq != nil {
+				instance.Disks = append(instance.Disks, &compute.AttachedDisk{DeviceName: disk1})
+			}
+			fcp.InsertInstance(instance, zone, node)
+			gceDriver := initGCEDriverWithCloudProvider(t, fcpBlocking)
+			cs := gceDriver.cs
+			for _, e := range tc.initialCache {
+				cs.opsManager.opsCache.InstanceOps.AddOp(e.Key, e.Op)
+			}
+			cs.opsManager.ready = true
+
+			runPublishRequest := func(req *csi.ControllerPublishVolumeRequest) <-chan error {
+				response := make(chan error)
+				go func() {
+					_, err := cs.ControllerPublishVolume(context.Background(), req)
+					response <- err
+				}()
+				return response
+			}
+			runUnpublishRequest := func(req *csi.ControllerUnpublishVolumeRequest) <-chan error {
+				response := make(chan error)
+				go func() {
+					_, err := cs.ControllerUnpublishVolume(context.Background(), req)
+					response <- err
+				}()
+				return response
+			}
+
+			var resp <-chan error
+			if tc.pubReq != nil {
+				resp = runPublishRequest(tc.pubReq)
+			} else if tc.unpubReq != nil {
+				resp = runUnpublishRequest(tc.unpubReq)
+			} else {
+				t.Errorf("invalid test case")
+			}
+
+			// If a key corresponding to the disk+instance found in the cache, controller will check the op status.
+			for _, v := range tc.checkOpStatusError {
+				s := gcecloudprovider.Signal{}
+				if v {
+					s.ReportError = true
+				}
+				execute := <-readyToExecute
+				execute <- s
+			}
+
+			// If a key corresponding to the disk+instance found in the cache, controller will check the op status.
+			for _, v := range tc.checkOpStatusRunning {
+				s := gcecloudprovider.Signal{}
+				if v {
+					s.ReportRunning = true
+				}
+				execute := <-readyToExecute
+				execute <- s
+			}
+
+			if tc.expectAttachDetachOpInCache {
+				// Find the running attach/detach op. This may need a few retries, because, at this time the controller publish/unpublish op will the cache and add a new op to cache.
+				backoff := wait.Backoff{
+					Duration: 10 * time.Millisecond,
+					Steps:    100,
+				}
+				if err := retry.OnError(backoff, func(err error) bool { return true }, func() error {
+					ops, innererr := cs.CloudProvider.ListZonalOps(context.Background(), map[string]bool{
+						"attachDisk": true, "detachDisk": true})
+					if innererr != nil {
+						return innererr
+					}
+					if len(ops) > 0 {
+						return nil
+					}
+					return fmt.Errorf("failed to find ops for fake cloud provider")
+				}); err != nil {
+					t.Errorf("timed out waiting for attach op to be updated")
+					return
+				}
+				ops, err := cs.CloudProvider.ListZonalOps(context.Background(), map[string]bool{"attachDisk": true})
+				if err != nil {
+					t.Errorf("Unexpected error in finding ops")
+				}
+				if len(ops) != 1 {
+					t.Errorf("Unexpected number of attach ops in cache")
+					return
+				}
+				// Now the controller has updated the cache and should block at poll
+				// verify cache content
+				opinfo := common.OpInfo{
+					Name: ops[0].Name,
+					Type: ops[0].OperationType,
+				}
+				if !containsDiskInstanceEntry(cs.opsManager.opsCache, DiskInstanceOpCacheEntry{Key: common.CreateDiskInstanceKey(project, zone, disk1, node), Op: opinfo}) {
+					t.Errorf("Unexpected cache entry detected")
+				}
+			}
+
+			// Unblock the poll operation
+			if tc.signalPollOp {
+				s := gcecloudprovider.Signal{}
+				execute := <-readyToExecute
+				execute <- s
+			}
+
+			err = <-resp
+			if tc.expectCSIOpError && err == nil {
+				t.Errorf("Expected error found none")
+			}
+			if !tc.expectCSIOpError && err != nil {
+				t.Errorf("Unexpected error found")
+			}
+		})
+	}
+}
+
+func TestHydrateCache(t *testing.T) {
+	containsOp := func(key common.InstanceKey, op common.OpInfo, cache *common.OpsCache) bool {
+		ops := cache.InstanceOps.GetOps(key)
+		for _, o := range ops {
+			if o.Name == op.Name && o.Type == op.Type {
+				return true
+			}
+		}
+		return false
+	}
+	tests := []struct {
+		name                 string
+		initialOps           []*compute.Operation
+		expectedCacheEntries []InstanceOpCacheEntry
+	}{
+		{
+			name: "no attach detach ops, no entries expected in cache",
+			initialOps: []*compute.Operation{
+				{
+					Name:          "op-1",
+					OperationType: "create",
+					Status:        "DONE",
+				},
+				{
+					Name:          "op-2",
+					OperationType: "create",
+					Status:        "DONE",
+				},
+			},
+		},
+		{
+			name: "done attach detach ops, no entries expected in cache",
+			initialOps: []*compute.Operation{
+				{
+					Name:          "op-1",
+					OperationType: "attachDisk",
+					Status:        "DONE",
+				},
+				{
+					Name:          "op-2",
+					OperationType: "detachDisk",
+					Status:        "DONE",
+				},
+			},
+		},
+		{
+			name: "in progress attach detach ops, entries expected in cache",
+			initialOps: []*compute.Operation{
+				{
+					Name:          "op-1",
+					OperationType: "attachDisk",
+					Status:        "PENDING",
+					TargetLink:    "https://www.googleapis.com/compute/v1/projects/testproject/zones/testzone/instances/testinstance",
+				},
+				{
+					Name:          "op-2",
+					OperationType: "detachDisk",
+					Status:        "RUNNING",
+					TargetLink:    "https://www.googleapis.com/compute/v1beta1/projects/testproject/zones/testzone/instances/testinstance",
+				},
+			},
+			expectedCacheEntries: []InstanceOpCacheEntry{
+				{
+					Key: "testproject_testzone_testinstance",
+					Op: common.OpInfo{
+						Name: "op-1",
+						Type: "attachDisk",
+					},
+				},
+				{
+					Key: "testproject_testzone_testinstance",
+					Op: common.OpInfo{
+						Name: "op-2",
+						Type: "detachDisk",
+					},
+				},
+			},
+		},
+		{
+			name: "in progress attach detach ops for multiple instances, entries expected in cache",
+			initialOps: []*compute.Operation{
+				{
+					Name:          "op-1",
+					OperationType: "attachDisk",
+					Status:        "PENDING",
+					TargetLink:    "https://www.googleapis.com/compute/v1/projects/testproject/zones/testzone/instances/testinstance",
+				},
+				{
+					Name:          "op-2",
+					OperationType: "detachDisk",
+					Status:        "RUNNING",
+					TargetLink:    "https://www.googleapis.com/compute/v1beta1/projects/testproject/zones/testzone/instances/testinstance",
+				},
+				{
+					Name:          "op-3",
+					OperationType: "attachDisk",
+					Status:        "PENDING",
+					TargetLink:    "https://www.googleapis.com/compute/v1/projects/testproject/zones/testzone/instances/testinstance1",
+				},
+				{
+					Name:          "op-4",
+					OperationType: "detachDisk",
+					Status:        "RUNNING",
+					TargetLink:    "https://www.googleapis.com/compute/v1beta1/projects/testproject/zones/testzone/instances/testinstance1",
+				},
+			},
+			expectedCacheEntries: []InstanceOpCacheEntry{
+				{
+					Key: "testproject_testzone_testinstance",
+					Op: common.OpInfo{
+						Name: "op-1",
+						Type: "attachDisk",
+					},
+				},
+				{
+					Key: "testproject_testzone_testinstance",
+					Op: common.OpInfo{
+						Name: "op-2",
+						Type: "detachDisk",
+					},
+				},
+				{
+					Key: "testproject_testzone_testinstance1",
+					Op: common.OpInfo{
+						Name: "op-3",
+						Type: "attachDisk",
+					},
+				},
+				{
+					Key: "testproject_testzone_testinstance1",
+					Op: common.OpInfo{
+						Name: "op-4",
+						Type: "detachDisk",
+					},
+				},
+			},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			fcp, err := gce.CreateFakeCloudProvider(project, zone, nil)
+			if err != nil {
+				t.Errorf("Failed to create fake cloud provider: %v", err)
+			}
+			fcp.InsertOps(tc.initialOps)
+			gceDriver := initGCEDriverWithCloudProvider(t, fcp)
+			gceDriver.cs.opsManager.HydrateOpsCache()
+			if !gceDriver.cs.opsManager.IsReady() {
+				t.Errorf("failed to initialize cache")
+			}
+			for _, e := range tc.expectedCacheEntries {
+				if !containsOp(e.Key, e.Op, gceDriver.cs.opsManager.opsCache) {
+					t.Errorf("expected entry not found")
+				}
+			}
+		})
+	}
+}
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 d645907e..a62e50dc 100644
--- a/pkg/gce-pd-csi-driver/gce-pd-driver_test.go
+++ b/pkg/gce-pd-csi-driver/gce-pd-driver_test.go
@@ -18,6 +18,7 @@ import (
 	"testing"
 
 	gce "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute"
+	gcecloudprovider "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute"
 )
 
 func initGCEDriver(t *testing.T, cloudDisks []*gce.CloudDisk) *GCEDriver {
@@ -28,7 +29,7 @@ func initGCEDriver(t *testing.T, cloudDisks []*gce.CloudDisk) *GCEDriver {
 	return initGCEDriverWithCloudProvider(t, fakeCloudProvider)
 }
 
-func initBlockingGCEDriver(t *testing.T, cloudDisks []*gce.CloudDisk, readyToExecute chan chan struct{}) *GCEDriver {
+func initBlockingGCEDriver(t *testing.T, cloudDisks []*gce.CloudDisk, readyToExecute chan chan gcecloudprovider.Signal) *GCEDriver {
 	fakeCloudProvider, err := gce.CreateFakeCloudProvider(project, zone, cloudDisks)
 	if err != nil {
 		t.Fatalf("Failed to create fake cloud provider: %v", err)