Repository: kubernetes-sigs/gcp-compute-persistent-disk-csi-driver
Pull request #1240 (2023-05-26 17:37:44)

Automated cherry pick of #1150 #1232 #1227 on release-1.9.
Cherry pick of #1150 #1232 #1227 on release-1.9.

#1150: satisfy volume cloning topology requirements when choosing zone for CreateVolume
#1232: Use errors.As so we can detect wrapped errors, and check for
#1227: Adding new metric pdcsi_operation_errors to fetch error

For details on the cherry pick process, see the [cherry pick requests](https://git.k8s.io/community/contributors/devel/sig-release/cherry-picks.md) page.

```release-note
#1150: satisfy volume cloning topology requirements when choosing zone for CreateVolume

#1232: Use errors.As so we can detect wrapped errors

#1227: Adding new metric pdcsi_operation_errors to fetch error
```


Patch:
diff --git a/cmd/gce-pd-csi-driver/main.go b/cmd/gce-pd-csi-driver/main.go
index 94836438..cffd1641 100644
--- a/cmd/gce-pd-csi-driver/main.go
+++ b/cmd/gce-pd-csi-driver/main.go
@@ -96,10 +96,14 @@ func handle() {
 	}
 	klog.V(2).Infof("Driver vendor version %v", version)
 
-	if *runControllerService && *httpEndpoint != "" && metrics.IsGKEComponentVersionAvailable() {
+	if *runControllerService && *httpEndpoint != "" {
 		mm := metrics.NewMetricsManager()
 		mm.InitializeHttpHandler(*httpEndpoint, *metricsPath)
-		mm.EmitGKEComponentVersion()
+		mm.RegisterPDCSIMetric()
+
+		if metrics.IsGKEComponentVersionAvailable() {
+			mm.EmitGKEComponentVersion()
+		}
 	}
 
 	if len(*extraVolumeLabelsStr) > 0 && !*runControllerService {
diff --git a/pkg/common/utils.go b/pkg/common/utils.go
index 3f2d9830..0c598d39 100644
--- a/pkg/common/utils.go
+++ b/pkg/common/utils.go
@@ -17,14 +17,21 @@ limitations under the License.
 package common
 
 import (
+	"context"
+	"errors"
 	"fmt"
+	"net/http"
 	"regexp"
 	"strings"
 
 	"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud/meta"
+	"google.golang.org/api/googleapi"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/apimachinery/pkg/util/sets"
 	volumehelpers "k8s.io/cloud-provider/volume/helpers"
+	"k8s.io/klog/v2"
 )
 
 const (
@@ -288,3 +295,83 @@ func ParseMachineType(machineTypeUrl string) (string, error) {
 	}
 	return machineType[1], nil
 }
+
+// CodeForError returns a pointer to the grpc error code that maps to the http
+// error code for the passed in user googleapi error or context error. Returns
+// codes.Internal if the given error is not a googleapi error caused by the user.
+// The following http error codes are considered user errors:
+// (1) http 400 Bad Request, returns grpc InvalidArgument,
+// (2) http 403 Forbidden, returns grpc PermissionDenied,
+// (3) http 404 Not Found, returns grpc NotFound
+// (4) http 429 Too Many Requests, returns grpc ResourceExhausted
+// The following errors are considered context errors:
+// (1) "context deadline exceeded", returns grpc DeadlineExceeded,
+// (2) "context canceled", returns grpc Canceled
+func CodeForError(err error) *codes.Code {
+	if err == nil {
+		return nil
+	}
+
+	if errCode := existingErrorCode(err); errCode != nil {
+		return errCode
+	}
+	if code := isContextError(err); code != nil {
+		return code
+	}
+
+	internalErrorCode := codes.Internal
+	// Upwrap the error
+	var apiErr *googleapi.Error
+	if !errors.As(err, &apiErr) {
+		return &internalErrorCode
+	}
+
+	userErrors := map[int]codes.Code{
+		http.StatusForbidden:       codes.PermissionDenied,
+		http.StatusBadRequest:      codes.InvalidArgument,
+		http.StatusTooManyRequests: codes.ResourceExhausted,
+		http.StatusNotFound:        codes.NotFound,
+	}
+	if code, ok := userErrors[apiErr.Code]; ok {
+		return &code
+	}
+
+	return &internalErrorCode
+}
+
+// isContextError returns a pointer to the grpc error code DeadlineExceeded
+// if the passed in error contains the "context deadline exceeded" string and returns
+// the grpc error code Canceled if the error contains the "context canceled" string.
+func isContextError(err error) *codes.Code {
+	if err == nil {
+		return nil
+	}
+
+	errStr := err.Error()
+	if strings.Contains(errStr, context.DeadlineExceeded.Error()) {
+		return errCodePtr(codes.DeadlineExceeded)
+	}
+	if strings.Contains(errStr, context.Canceled.Error()) {
+		return errCodePtr(codes.Canceled)
+	}
+	return nil
+}
+
+func existingErrorCode(err error) *codes.Code {
+	if err == nil {
+		return nil
+	}
+	if status, ok := status.FromError(err); ok {
+		return errCodePtr(status.Code())
+	}
+	return nil
+}
+
+func errCodePtr(code codes.Code) *codes.Code {
+	return &code
+}
+
+func LoggedError(msg string, err error) error {
+	klog.Errorf(msg+"%v", err.Error())
+	return status.Errorf(*CodeForError(err), msg+"%v", err.Error())
+}
diff --git a/pkg/gce-cloud-provider/compute/gce.go b/pkg/gce-cloud-provider/compute/gce.go
index 57a2b569..1d9cb051 100644
--- a/pkg/gce-cloud-provider/compute/gce.go
+++ b/pkg/gce-cloud-provider/compute/gce.go
@@ -16,6 +16,7 @@ package gcecloudprovider
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net/http"
 	"os"
@@ -248,8 +249,8 @@ func getProjectAndZone(config *ConfigFile) (string, string, error) {
 // isGCEError returns true if given error is a googleapi.Error with given
 // reason (e.g. "resourceInUseByAnotherResource")
 func IsGCEError(err error, reason string) bool {
-	apiErr, ok := err.(*googleapi.Error)
-	if !ok {
+	var apiErr *googleapi.Error
+	if !errors.As(err, &apiErr) {
 		return false
 	}
 
diff --git a/pkg/gce-pd-csi-driver/controller.go b/pkg/gce-pd-csi-driver/controller.go
index 4830aa23..b2157960 100644
--- a/pkg/gce-pd-csi-driver/controller.go
+++ b/pkg/gce-pd-csi-driver/controller.go
@@ -21,6 +21,7 @@ import (
 	"math/rand"
 	"regexp"
 	"sort"
+	"strings"
 	"time"
 
 	"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud/meta"
@@ -33,14 +34,17 @@ import (
 	"k8s.io/apimachinery/pkg/util/uuid"
 	"k8s.io/client-go/util/flowcontrol"
 	"k8s.io/klog/v2"
+	"k8s.io/utils/strings/slices"
 
 	"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"
+	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/metrics"
 )
 
 type GCEControllerServer struct {
 	Driver        *GCEDriver
 	CloudProvider gce.GCECompute
+	Metrics       metrics.MetricsManager
 
 	disks []*compute.Disk
 	seen  map[string]int
@@ -99,6 +103,14 @@ type workItem struct {
 	unpublishReq *csi.ControllerUnpublishVolumeRequest
 }
 
+// locationRequirements are additional location topology requirements that must be respected when creating a volume.
+type locationRequirements struct {
+	srcVolRegion         string
+	srcVolZone           string
+	srcReplicationType   string
+	cloneReplicationType string
+}
+
 var _ csi.ControllerServer = &GCEControllerServer{}
 
 const (
@@ -111,7 +123,7 @@ const (
 
 	replicationTypeNone       = "none"
 	replicationTypeRegionalPD = "regional-pd"
-
+	diskNotFound              = ""
 	// The maximum number of entries that we can include in the
 	// ListVolumesResposne
 	// In reality, the limit here is 4MB (based on gRPC client response limits),
@@ -143,8 +155,52 @@ func isDiskReady(disk *gce.CloudDisk) (bool, error) {
 	return false, nil
 }
 
+// cloningLocationRequirements returns additional location requirements to be applied to the given create volume requests topology.
+// If the CreateVolumeRequest will use volume cloning, location requirements in compliance with the volume cloning limitations
+// will be returned: https://cloud.google.com/kubernetes-engine/docs/how-to/persistent-volumes/volume-cloning#limitations.
+func cloningLocationRequirements(req *csi.CreateVolumeRequest, cloneReplicationType string) (*locationRequirements, error) {
+	if !useVolumeCloning(req) {
+		return nil, nil
+	}
+	// If we are using volume cloning, this will be set.
+	volSrc := req.VolumeContentSource.GetVolume()
+	volSrcVolID := volSrc.GetVolumeId()
+
+	_, sourceVolKey, err := common.VolumeIDToKey(volSrcVolID)
+	if err != nil {
+		return nil, fmt.Errorf("volume ID is invalid: %w", err)
+	}
+
+	isZonalSrcVol := sourceVolKey.Type() == meta.Zonal
+	if isZonalSrcVol {
+		region, err := common.GetRegionFromZones([]string{sourceVolKey.Zone})
+		if err != nil {
+			return nil, fmt.Errorf("failed to get region from zones: %w", err)
+		}
+		sourceVolKey.Region = region
+	}
+
+	srcReplicationType := replicationTypeNone
+	if !isZonalSrcVol {
+		srcReplicationType = replicationTypeRegionalPD
+	}
+
+	return &locationRequirements{srcVolZone: sourceVolKey.Zone, srcVolRegion: sourceVolKey.Region, srcReplicationType: srcReplicationType, cloneReplicationType: cloneReplicationType}, nil
+}
+
+// useVolumeCloning returns true if the create volume request should be created with volume cloning.
+func useVolumeCloning(req *csi.CreateVolumeRequest) bool {
+	return req.VolumeContentSource != nil && req.VolumeContentSource.GetVolume() != nil
+}
+
 func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
 	var err error
+	diskTypeForMetric := ""
+	defer func() {
+		if err != nil {
+			gceCS.Metrics.RecordOperationErrorMetrics("CreateVolume", err, diskTypeForMetric)
+		}
+	}()
 	// Validate arguments
 	volumeCapabilities := req.GetVolumeCapabilities()
 	name := req.GetName()
@@ -169,6 +225,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 	// Apply Parameters (case-insensitive). We leave validation of
 	// the values to the cloud provider.
 	params, err := common.ExtractAndDefaultParameters(req.GetParameters(), gceCS.Driver.name, gceCS.Driver.extraVolumeLabels)
+	diskTypeForMetric = params.DiskType
 	if err != nil {
 		return nil, status.Errorf(codes.InvalidArgument, "failed to extract parameters: %v", err.Error())
 	}
@@ -178,12 +235,21 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 	if multiWriter {
 		gceAPIVersion = gce.GCEAPIVersionBeta
 	}
+
+	var locationTopReq *locationRequirements
+	if useVolumeCloning(req) {
+		locationTopReq, err = cloningLocationRequirements(req, params.ReplicationType)
+		if err != nil {
+			return nil, status.Errorf(codes.InvalidArgument, "failed to get location requirements: %v", err.Error())
+		}
+	}
+
 	// Determine the zone or zones+region of the disk
 	var zones []string
 	var volKey *meta.Key
 	switch params.ReplicationType {
 	case replicationTypeNone:
-		zones, err = pickZones(ctx, gceCS, req.GetAccessibilityRequirements(), 1)
+		zones, err = pickZones(ctx, gceCS, req.GetAccessibilityRequirements(), 1, locationTopReq)
 		if err != nil {
 			return nil, status.Errorf(codes.InvalidArgument, "CreateVolume failed to pick zones for disk: %v", err.Error())
 		}
@@ -193,7 +259,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 		volKey = meta.ZonalKey(name, zones[0])
 
 	case replicationTypeRegionalPD:
-		zones, err = pickZones(ctx, gceCS, req.GetAccessibilityRequirements(), 2)
+		zones, err = pickZones(ctx, gceCS, req.GetAccessibilityRequirements(), 2, locationTopReq)
 		if err != nil {
 			return nil, status.Errorf(codes.InvalidArgument, "CreateVolume failed to pick zones for disk: %v", err.Error())
 		}
@@ -208,7 +274,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 
 	volumeID, err := common.KeyToVolumeID(volKey, gceCS.CloudProvider.GetDefaultProject())
 	if err != nil {
-		return nil, LoggedError("Failed to convert volume key to volume ID: ", err)
+		return nil, common.LoggedError("Failed to convert volume key to volume ID: ", err)
 	}
 	if acquired := gceCS.volumeLocks.TryAcquire(volumeID); !acquired {
 		return nil, status.Errorf(codes.Aborted, common.VolumeOperationAlreadyExistsFmt, volumeID)
@@ -217,9 +283,10 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 
 	// Validate if disk already exists
 	existingDisk, err := gceCS.CloudProvider.GetDisk(ctx, gceCS.CloudProvider.GetDefaultProject(), volKey, gceAPIVersion)
+	diskTypeForMetric = metrics.GetDiskType(existingDisk)
 	if err != nil {
 		if !gce.IsGCEError(err, "notFound") {
-			return nil, LoggedError("CreateVolume unknown get disk error when validating: ", err)
+			return nil, common.LoggedError("CreateVolume, failed to getDisk when validating: ", err)
 		}
 	}
 	if err == nil {
@@ -234,7 +301,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 
 		ready, err := isDiskReady(existingDisk)
 		if err != nil {
-			return nil, LoggedError("CreateVolume disk "+volKey.String()+" had error checking ready status: ", err)
+			return nil, common.LoggedError("CreateVolume disk "+volKey.String()+" had error checking ready status: ", err)
 		}
 		if !ready {
 			return nil, status.Errorf(codes.Internal, "CreateVolume existing disk %v is not ready", volKey)
@@ -255,7 +322,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 			// Verify that snapshot exists
 			sl, err := gceCS.getSnapshotByID(ctx, snapshotID)
 			if err != nil {
-				return nil, LoggedError("CreateVolume failed to get snapshot "+snapshotID+": ", err)
+				return nil, common.LoggedError("CreateVolume failed to get snapshot "+snapshotID+": ", err)
 			} else if len(sl.Entries) == 0 {
 				return nil, status.Errorf(codes.NotFound, "CreateVolume source snapshot %s does not exist", snapshotID)
 			}
@@ -275,7 +342,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 				if gce.IsGCEError(err, "notFound") {
 					return nil, status.Errorf(codes.NotFound, "CreateVolume source volume %s does not exist", volumeContentSourceVolumeID)
 				} else {
-					return nil, LoggedError("CreateVolume unknown get disk error when validating: ", err)
+					return nil, common.LoggedError("CreateVolume, getDisk error when validating: ", err)
 				}
 			}
 
@@ -313,7 +380,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 			// Verify the source disk is ready.
 			ready, err := isDiskReady(diskFromSourceVolume)
 			if err != nil {
-				return nil, LoggedError("CreateVolume disk from source volume "+sourceVolKey.String()+"  had error checking ready status: ", err)
+				return nil, common.LoggedError("CreateVolume disk from source volume "+sourceVolKey.String()+"  had error checking ready status: ", err)
 			}
 			if !ready {
 				return nil, status.Errorf(codes.Internal, "CreateVolume disk from source volume %v is not ready", sourceVolKey)
@@ -334,7 +401,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 		}
 		disk, err = createSingleZoneDisk(ctx, gceCS.CloudProvider, name, zones, params, capacityRange, capBytes, snapshotID, volumeContentSourceVolumeID, multiWriter)
 		if err != nil {
-			return nil, LoggedError("CreateVolume failed to create single zonal disk "+name+": ", err)
+			return nil, common.LoggedError("CreateVolume failed to create single zonal disk "+name+": ", err)
 		}
 	case replicationTypeRegionalPD:
 		if len(zones) != 2 {
@@ -342,7 +409,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 		}
 		disk, err = createRegionalDisk(ctx, gceCS.CloudProvider, name, zones, params, capacityRange, capBytes, snapshotID, volumeContentSourceVolumeID, multiWriter)
 		if err != nil {
-			return nil, LoggedError("CreateVolume failed to create regional disk "+name+": ", err)
+			return nil, common.LoggedError("CreateVolume failed to create regional disk "+name+": ", err)
 		}
 	default:
 		return nil, status.Errorf(codes.InvalidArgument, "CreateVolume replication type '%s' is not supported", params.ReplicationType)
@@ -362,6 +429,13 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 }
 
 func (gceCS *GCEControllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVolumeRequest) (*csi.DeleteVolumeResponse, error) {
+	var err error
+	diskTypeForMetric := ""
+	defer func() {
+		if err != nil {
+			gceCS.Metrics.RecordOperationErrorMetrics("DeleteVolume", err, diskTypeForMetric)
+		}
+	}()
 	// Validate arguments
 	volumeID := req.GetVolumeId()
 	if len(volumeID) == 0 {
@@ -382,17 +456,18 @@ func (gceCS *GCEControllerServer) DeleteVolume(ctx context.Context, req *csi.Del
 			klog.Warningf("DeleteVolume treating volume as deleted because cannot find volume %v: %v", volumeID, err.Error())
 			return &csi.DeleteVolumeResponse{}, nil
 		}
-		return nil, LoggedError("DeleteVolume error repairing underspecified volume key: ", err)
+		return nil, common.LoggedError("DeleteVolume error repairing underspecified volume key: ", err)
 	}
 
 	if acquired := gceCS.volumeLocks.TryAcquire(volumeID); !acquired {
 		return nil, status.Errorf(codes.Aborted, common.VolumeOperationAlreadyExistsFmt, volumeID)
 	}
 	defer gceCS.volumeLocks.Release(volumeID)
-
+	disk, _ := gceCS.CloudProvider.GetDisk(ctx, project, volKey, gce.GCEAPIVersionV1)
+	diskTypeForMetric = metrics.GetDiskType(disk)
 	err = gceCS.CloudProvider.DeleteDisk(ctx, project, volKey)
 	if err != nil {
-		return nil, LoggedError("unknown Delete disk error: ", err)
+		return nil, common.LoggedError("Failed to delete disk: ", err)
 	}
 
 	klog.V(4).Infof("DeleteVolume succeeded for disk %v", volKey)
@@ -400,8 +475,15 @@ func (gceCS *GCEControllerServer) DeleteVolume(ctx context.Context, req *csi.Del
 }
 
 func (gceCS *GCEControllerServer) ControllerPublishVolume(ctx context.Context, req *csi.ControllerPublishVolumeRequest) (*csi.ControllerPublishVolumeResponse, error) {
+	var err error
+	diskTypeForMetric := ""
+	defer func() {
+		if err != nil {
+			gceCS.Metrics.RecordOperationErrorMetrics("ControllerPublishVolume", err, diskTypeForMetric)
+		}
+	}()
 	// Only valid requests will be accepted
-	_, _, err := gceCS.validateControllerPublishVolumeRequest(ctx, req)
+	_, _, err = gceCS.validateControllerPublishVolumeRequest(ctx, req)
 	if err != nil {
 		return nil, err
 	}
@@ -411,7 +493,7 @@ func (gceCS *GCEControllerServer) ControllerPublishVolume(ctx context.Context, r
 		return nil, status.Errorf(codes.Unavailable, "ControllerPublish not permitted on node %q due to backoff condition", req.NodeId)
 	}
 
-	resp, err := gceCS.executeControllerPublishVolume(ctx, req)
+	resp, err, diskTypeForMetric := gceCS.executeControllerPublishVolume(ctx, req)
 	if err != nil {
 		klog.Infof("For node %s adding backoff due to error for volume %s: %v", req.NodeId, req.VolumeId, err.Error())
 		gceCS.errorBackoff.next(backoffId)
@@ -459,11 +541,11 @@ func parseMachineType(machineTypeUrl string) string {
 	return machineType
 }
 
-func (gceCS *GCEControllerServer) executeControllerPublishVolume(ctx context.Context, req *csi.ControllerPublishVolumeRequest) (*csi.ControllerPublishVolumeResponse, error) {
+func (gceCS *GCEControllerServer) executeControllerPublishVolume(ctx context.Context, req *csi.ControllerPublishVolumeRequest) (*csi.ControllerPublishVolumeResponse, error, string) {
+	diskType := ""
 	project, volKey, err := gceCS.validateControllerPublishVolumeRequest(ctx, req)
-
 	if err != nil {
-		return nil, err
+		return nil, err, diskType
 	}
 
 	volumeID := req.GetVolumeId()
@@ -478,36 +560,36 @@ func (gceCS *GCEControllerServer) executeControllerPublishVolume(ctx context.Con
 	project, volKey, err = gceCS.CloudProvider.RepairUnderspecifiedVolumeKey(ctx, project, volKey)
 	if err != nil {
 		if gce.IsGCENotFoundError(err) {
-			return nil, status.Errorf(codes.NotFound, "ControllerPublishVolume could not find volume with ID %v: %v", volumeID, err.Error())
+			return nil, status.Errorf(codes.NotFound, "ControllerPublishVolume could not find volume with ID %v: %v", volumeID, err.Error()), diskType
 		}
-		return nil, LoggedError("ControllerPublishVolume error repairing underspecified volume key: ", err)
+		return nil, common.LoggedError("ControllerPublishVolume error repairing underspecified volume key: ", err), diskType
 	}
 
 	// Acquires the lock for the volume on that node only, because we need to support the ability
 	// to publish the same volume onto different nodes concurrently
 	lockingVolumeID := fmt.Sprintf("%s/%s", nodeID, volumeID)
 	if acquired := gceCS.volumeLocks.TryAcquire(lockingVolumeID); !acquired {
-		return nil, status.Errorf(codes.Aborted, common.VolumeOperationAlreadyExistsFmt, lockingVolumeID)
+		return nil, status.Errorf(codes.Aborted, common.VolumeOperationAlreadyExistsFmt, lockingVolumeID), diskType
 	}
 	defer gceCS.volumeLocks.Release(lockingVolumeID)
-
-	_, err = gceCS.CloudProvider.GetDisk(ctx, project, volKey, gce.GCEAPIVersionV1)
+	disk, err := gceCS.CloudProvider.GetDisk(ctx, project, volKey, gce.GCEAPIVersionV1)
+	diskType = metrics.GetDiskType(disk)
 	if err != nil {
 		if gce.IsGCENotFoundError(err) {
-			return nil, status.Errorf(codes.NotFound, "Could not find disk %v: %v", volKey.String(), err.Error())
+			return nil, status.Errorf(codes.NotFound, "Could not find disk %v: %v", volKey.String(), err.Error()), diskType
 		}
-		return nil, status.Errorf(codes.Internal, "Unknown get disk error: %v", err.Error())
+		return nil, status.Errorf(codes.Internal, "Failed to getDisk: %v", err.Error()), diskType
 	}
 	instanceZone, instanceName, err := common.NodeIDToZoneAndName(nodeID)
 	if err != nil {
-		return nil, status.Errorf(codes.NotFound, "could not split nodeID: %v", err.Error())
+		return nil, status.Errorf(codes.NotFound, "could not split nodeID: %v", err.Error()), diskType
 	}
 	instance, err := gceCS.CloudProvider.GetInstanceOrError(ctx, instanceZone, instanceName)
 	if err != nil {
 		if gce.IsGCENotFoundError(err) {
-			return nil, status.Errorf(codes.NotFound, "Could not find instance %v: %v", nodeID, err.Error())
+			return nil, status.Errorf(codes.NotFound, "Could not find instance %v: %v", nodeID, err.Error()), diskType
 		}
-		return nil, status.Errorf(codes.Internal, "Unknown get instance error: %v", err.Error())
+		return nil, status.Errorf(codes.Internal, "Failed to get instance: %v", err.Error()), diskType
 	}
 
 	readWrite := "READ_WRITE"
@@ -517,21 +599,21 @@ func (gceCS *GCEControllerServer) executeControllerPublishVolume(ctx context.Con
 
 	deviceName, err := common.GetDeviceName(volKey)
 	if err != nil {
-		return nil, status.Errorf(codes.Internal, "error getting device name: %v", err.Error())
+		return nil, status.Errorf(codes.Internal, "error getting device name: %v", err.Error()), diskType
 	}
 
 	attached, err := diskIsAttachedAndCompatible(deviceName, instance, volumeCapability, readWrite)
 	if err != nil {
-		return nil, status.Errorf(codes.AlreadyExists, "Disk %v already published to node %v but incompatible: %v", volKey.Name, nodeID, err.Error())
+		return nil, status.Errorf(codes.AlreadyExists, "Disk %v already published to node %v but incompatible: %v", volKey.Name, nodeID, err.Error()), diskType
 	}
 	if attached {
 		// Volume is attached to node. Success!
 		klog.V(4).Infof("ControllerPublishVolume succeeded for disk %v to instance %v, already attached.", volKey, nodeID)
-		return pubVolResp, nil
+		return pubVolResp, nil, diskType
 	}
 	instanceZone, instanceName, err = common.NodeIDToZoneAndName(nodeID)
 	if err != nil {
-		return nil, status.Errorf(codes.InvalidArgument, "could not split nodeID: %v", err.Error())
+		return nil, status.Errorf(codes.InvalidArgument, "could not split nodeID: %v", err.Error()), diskType
 	}
 	err = gceCS.CloudProvider.AttachDisk(ctx, project, volKey, readWrite, attachableDiskTypePersistent, instanceZone, instanceName)
 	if err != nil {
@@ -540,32 +622,39 @@ func (gceCS *GCEControllerServer) executeControllerPublishVolume(ctx context.Con
 			// If we encountered an UnsupportedDiskError, rewrite the error message to be more user friendly.
 			// The error message from GCE is phrased around disk create on VM creation, not runtime attach.
 			machineType := parseMachineType(instance.MachineType)
-			return nil, status.Errorf(codes.InvalidArgument, "'%s' is not a compatible disk type with the machine type %s, please review the GCP online documentation for available persistent disk options", udErr.DiskType, machineType)
+			return nil, status.Errorf(codes.InvalidArgument, "'%s' is not a compatible disk type with the machine type %s, please review the GCP online documentation for available persistent disk options", udErr.DiskType, machineType), diskType
 		}
-		return nil, status.Errorf(codes.Internal, "unknown Attach error: %v", err.Error())
+		return nil, status.Errorf(codes.Internal, "Failed to Attach: %v", err.Error()), diskType
 	}
 
 	err = gceCS.CloudProvider.WaitForAttach(ctx, project, volKey, instanceZone, instanceName)
 	if err != nil {
-		return nil, status.Errorf(codes.Internal, "unknown WaitForAttach error: %v", err.Error())
+		return nil, status.Errorf(codes.Internal, "Errored during WaitForAttach: %v", err.Error()), diskType
 	}
+
 	klog.V(4).Infof("ControllerPublishVolume succeeded for disk %v to instance %v", volKey, nodeID)
-	return pubVolResp, nil
+	return pubVolResp, nil, diskType
 }
 
 func (gceCS *GCEControllerServer) ControllerUnpublishVolume(ctx context.Context, req *csi.ControllerUnpublishVolumeRequest) (*csi.ControllerUnpublishVolumeResponse, error) {
-	// Only valid requests will be queued
-	_, _, err := gceCS.validateControllerUnpublishVolumeRequest(ctx, req)
+	var err error
+	diskTypeForMetric := ""
+	defer func() {
+		if err != nil {
+			gceCS.Metrics.RecordOperationErrorMetrics("ControllerUnpublishVolume", err, diskTypeForMetric)
+		}
+	}()
+	_, _, err = gceCS.validateControllerUnpublishVolumeRequest(ctx, req)
 	if err != nil {
 		return nil, err
 	}
-
+	err = status.Errorf(codes.InvalidArgument, "error message")
+	// Only valid requests will be queued
 	backoffId := gceCS.errorBackoff.backoffId(req.NodeId, req.VolumeId)
 	if gceCS.errorBackoff.blocking(backoffId) {
 		return nil, status.Errorf(codes.Unavailable, "ControllerUnpublish not permitted on node %q due to backoff condition", req.NodeId)
 	}
-
-	resp, err := gceCS.executeControllerUnpublishVolume(ctx, req)
+	resp, err, diskTypeForMetric := gceCS.executeControllerUnpublishVolume(ctx, req)
 	if err != nil {
 		klog.Infof("For node %s adding backoff due to error for volume %s", req.NodeId, req.VolumeId)
 		gceCS.errorBackoff.next(backoffId)
@@ -595,11 +684,12 @@ func (gceCS *GCEControllerServer) validateControllerUnpublishVolumeRequest(ctx c
 	return project, volKey, nil
 }
 
-func (gceCS *GCEControllerServer) executeControllerUnpublishVolume(ctx context.Context, req *csi.ControllerUnpublishVolumeRequest) (*csi.ControllerUnpublishVolumeResponse, error) {
+func (gceCS *GCEControllerServer) executeControllerUnpublishVolume(ctx context.Context, req *csi.ControllerUnpublishVolumeRequest) (*csi.ControllerUnpublishVolumeResponse, error, string) {
+	var diskType string
 	project, volKey, err := gceCS.validateControllerUnpublishVolumeRequest(ctx, req)
 
 	if err != nil {
-		return nil, err
+		return nil, err, diskType
 	}
 
 	volumeID := req.GetVolumeId()
@@ -608,36 +698,37 @@ func (gceCS *GCEControllerServer) executeControllerUnpublishVolume(ctx context.C
 	if err != nil {
 		if gce.IsGCENotFoundError(err) {
 			klog.Warningf("Treating volume %v as unpublished because it could not be found", volumeID)
-			return &csi.ControllerUnpublishVolumeResponse{}, nil
+			return &csi.ControllerUnpublishVolumeResponse{}, nil, diskType
 		}
-		return nil, LoggedError("ControllerUnpublishVolume error repairing underspecified volume key: ", err)
+		return nil, common.LoggedError("ControllerUnpublishVolume error repairing underspecified volume key: ", err), diskType
 	}
 
 	// Acquires the lock for the volume on that node only, because we need to support the ability
 	// to unpublish the same volume from different nodes concurrently
 	lockingVolumeID := fmt.Sprintf("%s/%s", nodeID, volumeID)
 	if acquired := gceCS.volumeLocks.TryAcquire(lockingVolumeID); !acquired {
-		return nil, status.Errorf(codes.Aborted, common.VolumeOperationAlreadyExistsFmt, lockingVolumeID)
+		return nil, status.Errorf(codes.Aborted, common.VolumeOperationAlreadyExistsFmt, lockingVolumeID), diskType
 	}
 	defer gceCS.volumeLocks.Release(lockingVolumeID)
-
+	diskToUnpublish, _ := gceCS.CloudProvider.GetDisk(ctx, project, volKey, gce.GCEAPIVersionV1)
+	diskType = metrics.GetDiskType(diskToUnpublish)
 	instanceZone, instanceName, err := common.NodeIDToZoneAndName(nodeID)
 	if err != nil {
-		return nil, status.Errorf(codes.InvalidArgument, "could not split nodeID: %v", err.Error())
+		return nil, status.Errorf(codes.InvalidArgument, "could not split nodeID: %v", err.Error()), diskType
 	}
 	instance, err := gceCS.CloudProvider.GetInstanceOrError(ctx, instanceZone, instanceName)
 	if err != nil {
 		if gce.IsGCENotFoundError(err) {
 			// Node not existing on GCE means that disk has been detached
 			klog.Warningf("Treating volume %v as unpublished because node %v could not be found", volKey.String(), instanceName)
-			return &csi.ControllerUnpublishVolumeResponse{}, nil
+			return &csi.ControllerUnpublishVolumeResponse{}, nil, diskType
 		}
-		return nil, status.Errorf(codes.Internal, "error getting instance: %v", err.Error())
+		return nil, status.Errorf(codes.Internal, "error getting instance: %v", err.Error()), diskType
 	}
 
 	deviceName, err := common.GetDeviceName(volKey)
 	if err != nil {
-		return nil, status.Errorf(codes.Internal, "error getting device name: %v", err.Error())
+		return nil, status.Errorf(codes.Internal, "error getting device name: %v", err.Error()), diskType
 	}
 
 	attached := diskIsAttached(deviceName, instance)
@@ -645,19 +736,25 @@ func (gceCS *GCEControllerServer) executeControllerUnpublishVolume(ctx context.C
 	if !attached {
 		// Volume is not attached to node. Success!
 		klog.V(4).Infof("ControllerUnpublishVolume succeeded for disk %v from node %v. Already not attached.", volKey, nodeID)
-		return &csi.ControllerUnpublishVolumeResponse{}, nil
+		return &csi.ControllerUnpublishVolumeResponse{}, nil, diskType
 	}
-
 	err = gceCS.CloudProvider.DetachDisk(ctx, project, deviceName, instanceZone, instanceName)
 	if err != nil {
-		return nil, LoggedError("unknown detach error: ", err)
+		return nil, common.LoggedError("Failed to detach: ", err), diskType
 	}
 
 	klog.V(4).Infof("ControllerUnpublishVolume succeeded for disk %v from node %v", volKey, nodeID)
-	return &csi.ControllerUnpublishVolumeResponse{}, nil
+	return &csi.ControllerUnpublishVolumeResponse{}, nil, diskType
 }
 
 func (gceCS *GCEControllerServer) ValidateVolumeCapabilities(ctx context.Context, req *csi.ValidateVolumeCapabilitiesRequest) (*csi.ValidateVolumeCapabilitiesResponse, error) {
+	var err error
+	diskTypeForMetric := ""
+	defer func() {
+		if err != nil {
+			gceCS.Metrics.RecordOperationErrorMetrics("ValidateVolumeCapabilities", err, diskTypeForMetric)
+		}
+	}()
 	if req.GetVolumeCapabilities() == nil || len(req.GetVolumeCapabilities()) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "Volume Capabilities must be provided")
 	}
@@ -669,13 +766,12 @@ func (gceCS *GCEControllerServer) ValidateVolumeCapabilities(ctx context.Context
 	if err != nil {
 		return nil, status.Errorf(codes.InvalidArgument, "Volume ID is invalid: %v", err.Error())
 	}
-
 	project, volKey, err = gceCS.CloudProvider.RepairUnderspecifiedVolumeKey(ctx, project, volKey)
 	if err != nil {
 		if gce.IsGCENotFoundError(err) {
 			return nil, status.Errorf(codes.NotFound, "ValidateVolumeCapabilities could not find volume with ID %v: %v", volumeID, err.Error())
 		}
-		return nil, LoggedError("ValidateVolumeCapabilities error repairing underspecified volume key: ", err)
+		return nil, common.LoggedError("ValidateVolumeCapabilities error repairing underspecified volume key: ", err)
 	}
 
 	if acquired := gceCS.volumeLocks.TryAcquire(volumeID); !acquired {
@@ -684,11 +780,12 @@ func (gceCS *GCEControllerServer) ValidateVolumeCapabilities(ctx context.Context
 	defer gceCS.volumeLocks.Release(volumeID)
 
 	disk, err := gceCS.CloudProvider.GetDisk(ctx, project, volKey, gce.GCEAPIVersionV1)
+	diskTypeForMetric = metrics.GetDiskType(disk)
 	if err != nil {
 		if gce.IsGCENotFoundError(err) {
 			return nil, status.Errorf(codes.NotFound, "Could not find disk %v: %v", volKey.Name, err.Error())
 		}
-		return nil, LoggedError("Unknown get disk error: ", err)
+		return nil, common.LoggedError("Failed to getDisk: ", err)
 	}
 
 	// Check Volume Context is Empty
@@ -751,7 +848,7 @@ func (gceCS *GCEControllerServer) ListVolumes(ctx context.Context, req *csi.List
 			if gce.IsGCEInvalidError(err) {
 				return nil, status.Errorf(codes.Aborted, "ListVolumes error with invalid request: %v", err.Error())
 			}
-			return nil, LoggedError("Unknown list disk error: ", err)
+			return nil, common.LoggedError("Failed to list disk: ", err)
 		}
 		gceCS.disks = diskList
 		gceCS.seen = map[string]int{}
@@ -810,6 +907,13 @@ func (gceCS *GCEControllerServer) ControllerGetCapabilities(ctx context.Context,
 }
 
 func (gceCS *GCEControllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateSnapshotRequest) (*csi.CreateSnapshotResponse, error) {
+	var err error
+	diskTypeForMetric := ""
+	defer func() {
+		if err != nil {
+			gceCS.Metrics.RecordOperationErrorMetrics("CreateSnapshot", err, diskTypeForMetric)
+		}
+	}()
 	// Validate arguments
 	volumeID := req.GetSourceVolumeId()
 	if len(req.Name) == 0 {
@@ -829,12 +933,13 @@ func (gceCS *GCEControllerServer) CreateSnapshot(ctx context.Context, req *csi.C
 	defer gceCS.volumeLocks.Release(volumeID)
 
 	// Check if volume exists
-	_, err = gceCS.CloudProvider.GetDisk(ctx, project, volKey, gce.GCEAPIVersionV1)
+	disk, err := gceCS.CloudProvider.GetDisk(ctx, project, volKey, gce.GCEAPIVersionV1)
+	diskTypeForMetric = metrics.GetDiskType(disk)
 	if err != nil {
 		if gce.IsGCENotFoundError(err) {
 			return nil, status.Errorf(codes.NotFound, "CreateSnapshot could not find disk %v: %v", volKey.String(), err.Error())
 		}
-		return nil, LoggedError("CreateSnapshot unknown get disk error: ", err)
+		return nil, common.LoggedError("CreateSnapshot, failed to getDisk: ", err)
 	}
 
 	snapshotParams, err := common.ExtractAndDefaultSnapshotParameters(req.GetParameters(), gceCS.Driver.name)
@@ -867,13 +972,12 @@ func (gceCS *GCEControllerServer) createPDSnapshot(ctx context.Context, project
 	if err != nil {
 		return nil, status.Errorf(codes.InvalidArgument, "Invalid volume key: %v", volKey)
 	}
-
 	// Check if PD snapshot already exists
 	var snapshot *compute.Snapshot
 	snapshot, err = gceCS.CloudProvider.GetSnapshot(ctx, project, snapshotName)
 	if err != nil {
 		if !gce.IsGCEError(err, "notFound") {
-			return nil, status.Errorf(codes.Internal, "Unknown get snapshot error: %v", err.Error())
+			return nil, status.Errorf(codes.Internal, "Failed to get snapshot: %v", err.Error())
 		}
 		// If we could not find the snapshot, we create a new one
 		snapshot, err = gceCS.CloudProvider.CreateSnapshot(ctx, project, volKey, snapshotName, snapshotParams)
@@ -881,7 +985,7 @@ func (gceCS *GCEControllerServer) createPDSnapshot(ctx context.Context, project
 			if gce.IsGCEError(err, "notFound") {
 				return nil, status.Errorf(codes.NotFound, "Could not find volume with ID %v: %v", volKey.String(), err.Error())
 			}
-			return nil, LoggedError("Unknown create snapshot error: ", err)
+			return nil, common.LoggedError("Failed to create snapshot: ", err)
 		}
 	}
 
@@ -920,7 +1024,7 @@ func (gceCS *GCEControllerServer) createImage(ctx context.Context, project strin
 	image, err = gceCS.CloudProvider.GetImage(ctx, project, imageName)
 	if err != nil {
 		if !gce.IsGCEError(err, "notFound") {
-			return nil, LoggedError("Unknown get image error: ", err)
+			return nil, common.LoggedError("Failed to get image: ", err)
 		}
 		// create a new image
 		image, err = gceCS.CloudProvider.CreateImage(ctx, project, volKey, imageName, snapshotParams)
@@ -928,7 +1032,7 @@ func (gceCS *GCEControllerServer) createImage(ctx context.Context, project strin
 			if gce.IsGCEError(err, "notFound") {
 				return nil, status.Errorf(codes.NotFound, "Could not find volume with ID %v: %v", volKey.String(), err.Error())
 			}
-			return nil, LoggedError("Unknown create image error: ", err)
+			return nil, common.LoggedError("Failed to create image: ", err)
 		}
 	}
 
@@ -1040,6 +1144,13 @@ func isCSISnapshotReady(status string) (bool, error) {
 }
 
 func (gceCS *GCEControllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteSnapshotRequest) (*csi.DeleteSnapshotResponse, error) {
+	var err error
+	diskTypeForMetric := ""
+	defer func() {
+		if err != nil {
+			gceCS.Metrics.RecordOperationErrorMetrics("DeleteSnapshot", err, diskTypeForMetric)
+		}
+	}()
 	// Validate arguments
 	snapshotID := req.GetSnapshotId()
 	if len(snapshotID) == 0 {
@@ -1058,12 +1169,12 @@ func (gceCS *GCEControllerServer) DeleteSnapshot(ctx context.Context, req *csi.D
 	case common.DiskSnapshotType:
 		err = gceCS.CloudProvider.DeleteSnapshot(ctx, project, key)
 		if err != nil {
-			return nil, LoggedError("unknown Delete snapshot error: ", err)
+			return nil, common.LoggedError("Failed to DeleteSnapshot: ", err)
 		}
 	case common.DiskImageType:
 		err = gceCS.CloudProvider.DeleteImage(ctx, project, key)
 		if err != nil {
-			return nil, LoggedError("unknown Delete image error: ", err)
+			return nil, common.LoggedError("Failed to DeleteImage error: ", err)
 		}
 	default:
 		return nil, status.Errorf(codes.InvalidArgument, "unknown snapshot type %s", snapshotType)
@@ -1095,7 +1206,7 @@ func (gceCS *GCEControllerServer) ListSnapshots(ctx context.Context, req *csi.Li
 			if gce.IsGCEInvalidError(err) {
 				return nil, status.Errorf(codes.Aborted, "ListSnapshots error with invalid request: %v", err.Error())
 			}
-			return nil, LoggedError("Unknown list snapshots error: ", err)
+			return nil, common.LoggedError("Failed to list snapshots: ", err)
 		}
 		gceCS.snapshots = snapshotList
 		gceCS.snapshotTokens = map[string]int{}
@@ -1124,6 +1235,13 @@ func (gceCS *GCEControllerServer) ListSnapshots(ctx context.Context, req *csi.Li
 }
 
 func (gceCS *GCEControllerServer) ControllerExpandVolume(ctx context.Context, req *csi.ControllerExpandVolumeRequest) (*csi.ControllerExpandVolumeResponse, error) {
+	var err error
+	diskTypeForMetric := ""
+	defer func() {
+		if err != nil {
+			gceCS.Metrics.RecordOperationErrorMetrics("ControllerExpandVolume", err, diskTypeForMetric)
+		}
+	}()
 	volumeID := req.GetVolumeId()
 	if len(volumeID) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "ControllerExpandVolume volume ID must be provided")
@@ -1138,18 +1256,18 @@ func (gceCS *GCEControllerServer) ControllerExpandVolume(ctx context.Context, re
 	if err != nil {
 		return nil, status.Errorf(codes.InvalidArgument, "ControllerExpandVolume Volume ID is invalid: %v", err.Error())
 	}
-
 	project, volKey, err = gceCS.CloudProvider.RepairUnderspecifiedVolumeKey(ctx, project, volKey)
 	if err != nil {
 		if gce.IsGCENotFoundError(err) {
 			return nil, status.Errorf(codes.NotFound, "ControllerExpandVolume could not find volume with ID %v: %v", volumeID, err.Error())
 		}
-		return nil, LoggedError("ControllerExpandVolume error repairing underspecified volume key: ", err)
+		return nil, common.LoggedError("ControllerExpandVolume error repairing underspecified volume key: ", err)
 	}
-
+	sourceDisk, err := gceCS.CloudProvider.GetDisk(ctx, project, volKey, gce.GCEAPIVersionV1)
+	diskTypeForMetric = metrics.GetDiskType(sourceDisk)
 	resizedGb, err := gceCS.CloudProvider.ResizeDisk(ctx, project, volKey, reqBytes)
 	if err != nil {
-		return nil, LoggedError("ControllerExpandVolume failed to resize disk: ", err)
+		return nil, common.LoggedError("ControllerExpandVolume failed to resize disk: ", err)
 	}
 
 	klog.V(4).Infof("ControllerExpandVolume succeeded for disk %v to size %v", volKey, resizedGb)
@@ -1172,7 +1290,7 @@ func (gceCS *GCEControllerServer) getSnapshots(ctx context.Context, req *csi.Lis
 		if gce.IsGCEError(err, "invalid") {
 			return nil, status.Errorf(codes.Aborted, "Invalid error: %v", err.Error())
 		}
-		return nil, LoggedError("Unknown list snapshot error: ", err)
+		return nil, common.LoggedError("Failed to list snapshot: ", err)
 	}
 
 	images, _, err = gceCS.CloudProvider.ListImages(ctx, filter)
@@ -1180,7 +1298,7 @@ func (gceCS *GCEControllerServer) getSnapshots(ctx context.Context, req *csi.Lis
 		if gce.IsGCEError(err, "invalid") {
 			return nil, status.Errorf(codes.Aborted, "Invalid error: %v", err.Error())
 		}
-		return nil, LoggedError("Unknown list image error: ", err)
+		return nil, common.LoggedError("Failed to list image: ", err)
 	}
 
 	entries := []*csi.ListSnapshotsResponse_Entry{}
@@ -1221,7 +1339,7 @@ func (gceCS *GCEControllerServer) getSnapshotByID(ctx context.Context, snapshotI
 				// return empty list if no snapshot is found
 				return &csi.ListSnapshotsResponse{}, nil
 			}
-			return nil, LoggedError("Unknown list snapshot error: ", err)
+			return nil, common.LoggedError("Failed to list snapshot: ", err)
 		}
 		e, err := generateDiskSnapshotEntry(snapshot)
 		if err != nil {
@@ -1375,7 +1493,29 @@ func diskIsAttachedAndCompatible(deviceName string, instance *compute.Instance,
 	return false, nil
 }
 
-func pickZonesFromTopology(top *csi.TopologyRequirement, numZones int) ([]string, error) {
+// pickZonesInRegion will remove any zones that are not in the given region.
+func pickZonesInRegion(region string, zones []string) []string {
+	refinedZones := []string{}
+	for _, zone := range zones {
+		if strings.Contains(zone, region) {
+			refinedZones = append(refinedZones, zone)
+		}
+	}
+	return refinedZones
+}
+
+func prependZone(zone string, zones []string) []string {
+	newZones := []string{zone}
+	for i := 0; i < len(zones); i++ {
+		// Do not add a zone if it is equal to the zone that is already prepended to newZones.
+		if zones[i] != zone {
+			newZones = append(newZones, zones[i])
+		}
+	}
+	return newZones
+}
+
+func pickZonesFromTopology(top *csi.TopologyRequirement, numZones int, locationTopReq *locationRequirements) ([]string, error) {
 	reqZones, err := getZonesFromTopology(top.GetRequisite())
 	if err != nil {
 		return nil, fmt.Errorf("could not get zones from requisite topology: %w", err)
@@ -1385,6 +1525,39 @@ func pickZonesFromTopology(top *csi.TopologyRequirement, numZones int) ([]string
 		return nil, fmt.Errorf("could not get zones from preferred topology: %w", err)
 	}
 
+	if locationTopReq != nil {
+		srcVolZone := locationTopReq.srcVolZone
+		switch locationTopReq.cloneReplicationType {
+		// For zonal -> zonal cloning, the source disk zone must match the destination disk zone.
+		case replicationTypeNone:
+			// If the source volume zone is not in the topology requirement, we return an error.
+			if !slices.Contains(prefZones, srcVolZone) && !slices.Contains(reqZones, srcVolZone) {
+				volumeCloningReq := fmt.Sprintf("clone zone must match source disk zone: %s", srcVolZone)
+				return nil, fmt.Errorf("failed to find zone from topology %v: %s", top, volumeCloningReq)
+			}
+			return []string{srcVolZone}, nil
+		// For zonal or regional -> regional disk cloning, the source disk region must match the destination disk region.
+		case replicationTypeRegionalPD:
+			srcVolRegion := locationTopReq.srcVolRegion
+			prefZones = pickZonesInRegion(srcVolRegion, prefZones)
+			reqZones = pickZonesInRegion(srcVolRegion, reqZones)
+
+			if len(prefZones) == 0 && len(reqZones) == 0 {
+				volumeCloningReq := fmt.Sprintf("clone zone must reside in source disk region %s", srcVolRegion)
+				return nil, fmt.Errorf("failed to find zone from topology %v: %s", top, volumeCloningReq)
+			}
+
+			// For zonal -> regional disk cloning, one of the replicated zones must match the source zone.
+			if locationTopReq.srcReplicationType == replicationTypeNone {
+				if !slices.Contains(prefZones, srcVolZone) && !slices.Contains(reqZones, srcVolZone) {
+					volumeCloningReq := fmt.Sprintf("one of the replica zones of the clone must match the source disk zone: %s", srcVolZone)
+					return nil, fmt.Errorf("failed to find zone from topology %v: %s", top, volumeCloningReq)
+				}
+				prefZones = prependZone(srcVolZone, prefZones)
+			}
+		}
+	}
+
 	if numZones <= len(prefZones) {
 		return prefZones[0:numZones], nil
 	} else {
@@ -1443,16 +1616,25 @@ func getZoneFromSegment(seg map[string]string) (string, error) {
 	return zone, nil
 }
 
-func pickZones(ctx context.Context, gceCS *GCEControllerServer, top *csi.TopologyRequirement, numZones int) ([]string, error) {
+func pickZones(ctx context.Context, gceCS *GCEControllerServer, top *csi.TopologyRequirement, numZones int, locationTopReq *locationRequirements) ([]string, error) {
 	var zones []string
 	var err error
 	if top != nil {
-		zones, err = pickZonesFromTopology(top, numZones)
+		zones, err = pickZonesFromTopology(top, numZones, locationTopReq)
 		if err != nil {
 			return nil, fmt.Errorf("failed to pick zones from topology: %w", err)
 		}
 	} else {
-		zones, err = getDefaultZonesInRegion(ctx, gceCS, []string{gceCS.CloudProvider.GetDefaultZone()}, numZones)
+		existingZones := []string{gceCS.CloudProvider.GetDefaultZone()}
+		// We set existingZones to the source volume zone so that for zonal -> zonal cloning, the clone is provisioned
+		// in the same zone as the source volume, and for zonal -> regional, one of the replicated zones will always
+		// be the zone of the source volume. For regional -> regional cloning, the srcVolZone will not be set, so we
+		// just use the default zone.
+		if locationTopReq != nil && locationTopReq.srcReplicationType == replicationTypeNone {
+			existingZones = []string{locationTopReq.srcVolZone}
+		}
+		// If topology is nil, then the Immediate binding mode was used without setting allowedTopologies in the storageclass.
+		zones, err = getDefaultZonesInRegion(ctx, gceCS, existingZones, numZones)
 		if err != nil {
 			return nil, fmt.Errorf("failed to get default %v zones in region: %w", numZones, err)
 		}
diff --git a/pkg/gce-pd-csi-driver/utils.go b/pkg/gce-pd-csi-driver/utils.go
index c81fc5be..d8ef8ce0 100644
--- a/pkg/gce-pd-csi-driver/utils.go
+++ b/pkg/gce-pd-csi-driver/utils.go
@@ -17,17 +17,12 @@ limitations under the License.
 package gceGCEDriver
 
 import (
+	"context"
 	"errors"
 	"fmt"
-	"net/http"
-
-	"context"
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
-	"google.golang.org/api/googleapi"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
 	"k8s.io/klog/v2"
 )
 
@@ -217,36 +212,3 @@ func containsZone(zones []string, zone string) bool {
 
 	return false
 }
-
-// CodeForError returns a pointer to the grpc error code that maps to the http
-// error code for the passed in user googleapi error. Returns codes.Internal if
-// the given error is not a googleapi error caused by the user. The following
-// http error codes are considered user errors:
-// (1) http 400 Bad Request, returns grpc InvalidArgument,
-// (2) http 403 Forbidden, returns grpc PermissionDenied,
-// (3) http 404 Not Found, returns grpc NotFound
-// (4) http 429 Too Many Requests, returns grpc ResourceExhausted
-func CodeForError(err error) *codes.Code {
-	internalErrorCode := codes.Internal
-	// Upwrap the error
-	var apiErr *googleapi.Error
-	if !errors.As(err, &apiErr) {
-		return &internalErrorCode
-	}
-
-	userErrors := map[int]codes.Code{
-		http.StatusForbidden:       codes.PermissionDenied,
-		http.StatusBadRequest:      codes.InvalidArgument,
-		http.StatusTooManyRequests: codes.ResourceExhausted,
-		http.StatusNotFound:        codes.NotFound,
-	}
-	if code, ok := userErrors[apiErr.Code]; ok {
-		return &code
-	}
-	return &internalErrorCode
-}
-
-func LoggedError(msg string, err error) error {
-	klog.Errorf(msg+"%v", err.Error())
-	return status.Errorf(*CodeForError(err), msg+"%v", err.Error())
-}
diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go
index cae37923..43282ee4 100644
--- a/pkg/metrics/metrics.go
+++ b/pkg/metrics/metrics.go
@@ -23,12 +23,15 @@ import (
 
 	"k8s.io/component-base/metrics"
 	"k8s.io/klog/v2"
+	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
+	gce "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute"
 )
 
 const (
 	// envGKEPDCSIVersion is an environment variable set in the PDCSI controller manifest
 	// with the current version of the GKE component.
 	envGKEPDCSIVersion = "GKE_PDCSI_VERSION"
+	pdcsiDriverName    = "pd.csi.storage.gke.io"
 )
 
 var (
@@ -37,28 +40,41 @@ var (
 		Name: "component_version",
 		Help: "Metric to expose the version of the PDCSI GKE component.",
 	}, []string{"component_version"})
+
+	pdcsiOperationErrorsMetric = metrics.NewCounterVec(
+		&metrics.CounterOpts{
+			Subsystem:      "csidriver",
+			Name:           "operation_errors",
+			Help:           "CSI server side error metrics",
+			StabilityLevel: metrics.ALPHA,
+		},
+		[]string{"driver_name", "method_name", "grpc_status_code", "disk_type"})
 )
 
-type metricsManager struct {
+type MetricsManager struct {
 	registry metrics.KubeRegistry
 }
 
-func NewMetricsManager() metricsManager {
-	mm := metricsManager{
+func NewMetricsManager() MetricsManager {
+	mm := MetricsManager{
 		registry: metrics.NewKubeRegistry(),
 	}
 	return mm
 }
 
-func (mm *metricsManager) GetRegistry() metrics.KubeRegistry {
+func (mm *MetricsManager) GetRegistry() metrics.KubeRegistry {
 	return mm.registry
 }
 
-func (mm *metricsManager) registerComponentVersionMetric() {
+func (mm *MetricsManager) registerComponentVersionMetric() {
 	mm.registry.MustRegister(gkeComponentVersion)
 }
 
-func (mm *metricsManager) recordComponentVersionMetric() error {
+func (mm *MetricsManager) RegisterPDCSIMetric() {
+	mm.registry.MustRegister(pdcsiOperationErrorsMetric)
+}
+
+func (mm *MetricsManager) recordComponentVersionMetric() error {
 	v := getEnvVar(envGKEPDCSIVersion)
 	if v == "" {
 		klog.V(2).Info("Skip emitting component version metric")
@@ -70,7 +86,14 @@ func (mm *metricsManager) recordComponentVersionMetric() error {
 	return nil
 }
 
-func (mm *metricsManager) EmitGKEComponentVersion() error {
+func (mm *MetricsManager) RecordOperationErrorMetrics(
+	operationName string,
+	operationErr error,
+	diskType string) {
+	pdcsiOperationErrorsMetric.WithLabelValues(pdcsiDriverName, "/csi.v1.Controller/"+operationName, common.CodeForError(operationErr).String(), diskType).Inc()
+}
+
+func (mm *MetricsManager) EmitGKEComponentVersion() error {
 	mm.registerComponentVersionMetric()
 	if err := mm.recordComponentVersionMetric(); err != nil {
 		return err
@@ -87,7 +110,7 @@ type Server interface {
 
 // RegisterToServer registers an HTTP handler for this metrics manager to the
 // given server at the specified address/path.
-func (mm *metricsManager) registerToServer(s Server, metricsPath string) {
+func (mm *MetricsManager) registerToServer(s Server, metricsPath string) {
 	s.Handle(metricsPath, metrics.HandlerFor(
 		mm.GetRegistry(),
 		metrics.HandlerOpts{
@@ -95,7 +118,7 @@ func (mm *metricsManager) registerToServer(s Server, metricsPath string) {
 }
 
 // InitializeHttpHandler sets up a server and creates a handler for metrics.
-func (mm *metricsManager) InitializeHttpHandler(address, path string) {
+func (mm *MetricsManager) InitializeHttpHandler(address, path string) {
 	mux := http.NewServeMux()
 	mm.registerToServer(mux, path)
 	go func() {
@@ -123,3 +146,11 @@ func IsGKEComponentVersionAvailable() bool {
 
 	return true
 }
+
+func GetDiskType(disk *gce.CloudDisk) string {
+	var diskType string
+	if disk != nil {
+		diskType = disk.GetPDType()
+	}
+	return diskType
+}


Test patch:
diff --git a/pkg/common/utils_test.go b/pkg/common/utils_test.go
index a888465d..7fd31305 100644
--- a/pkg/common/utils_test.go
+++ b/pkg/common/utils_test.go
@@ -17,11 +17,17 @@ limitations under the License.
 package common
 
 import (
+	"context"
+	"errors"
 	"fmt"
+	"net/http"
 	"reflect"
 	"testing"
 
 	"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud/meta"
+	"google.golang.org/api/googleapi"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 const (
@@ -841,3 +847,109 @@ func TestParseMachineType(t *testing.T) {
 		})
 	}
 }
+
+func TestCodeForError(t *testing.T) {
+	internalErrorCode := codes.Internal
+	userErrorCode := codes.InvalidArgument
+	testCases := []struct {
+		name     string
+		inputErr error
+		expCode  *codes.Code
+	}{
+		{
+			name:     "Not googleapi.Error",
+			inputErr: errors.New("I am not a googleapi.Error"),
+			expCode:  &internalErrorCode,
+		},
+		{
+			name:     "User error",
+			inputErr: &googleapi.Error{Code: http.StatusBadRequest, Message: "User error with bad request"},
+			expCode:  &userErrorCode,
+		},
+		{
+			name:     "googleapi.Error but not a user error",
+			inputErr: &googleapi.Error{Code: http.StatusInternalServerError, Message: "Internal error"},
+			expCode:  &internalErrorCode,
+		},
+		{
+			name:     "context canceled error",
+			inputErr: context.Canceled,
+			expCode:  errCodePtr(codes.Canceled),
+		},
+		{
+			name:     "context deadline exceeded error",
+			inputErr: context.DeadlineExceeded,
+			expCode:  errCodePtr(codes.DeadlineExceeded),
+		},
+		{
+			name:     "status error with Aborted error code",
+			inputErr: status.Error(codes.Aborted, "aborted error"),
+			expCode:  errCodePtr(codes.Aborted),
+		},
+		{
+			name:     "nil error",
+			inputErr: nil,
+			expCode:  nil,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Logf("Running test: %v", tc.name)
+		errCode := CodeForError(tc.inputErr)
+		if (tc.expCode == nil) != (errCode == nil) {
+			t.Errorf("test %v failed: got %v, expected %v", tc.name, errCode, tc.expCode)
+		}
+		if tc.expCode != nil && *errCode != *tc.expCode {
+			t.Errorf("test %v failed: got %v, expected %v", tc.name, errCode, tc.expCode)
+		}
+	}
+}
+
+func TestIsContextError(t *testing.T) {
+	cases := []struct {
+		name            string
+		err             error
+		expectedErrCode *codes.Code
+	}{
+		{
+			name:            "deadline exceeded error",
+			err:             context.DeadlineExceeded,
+			expectedErrCode: errCodePtr(codes.DeadlineExceeded),
+		},
+		{
+			name:            "contains 'context deadline exceeded'",
+			err:             fmt.Errorf("got error: %w", context.DeadlineExceeded),
+			expectedErrCode: errCodePtr(codes.DeadlineExceeded),
+		},
+		{
+			name:            "context canceled error",
+			err:             context.Canceled,
+			expectedErrCode: errCodePtr(codes.Canceled),
+		},
+		{
+			name:            "contains 'context canceled'",
+			err:             fmt.Errorf("got error: %w", context.Canceled),
+			expectedErrCode: errCodePtr(codes.Canceled),
+		},
+		{
+			name:            "does not contain 'context canceled' or 'context deadline exceeded'",
+			err:             fmt.Errorf("unknown error"),
+			expectedErrCode: nil,
+		},
+		{
+			name:            "nil error",
+			err:             nil,
+			expectedErrCode: nil,
+		},
+	}
+
+	for _, test := range cases {
+		errCode := isContextError(test.err)
+		if (test.expectedErrCode == nil) != (errCode == nil) {
+			t.Errorf("test %v failed: got %v, expected %v", test.name, errCode, test.expectedErrCode)
+		}
+		if test.expectedErrCode != nil && *errCode != *test.expectedErrCode {
+			t.Errorf("test %v failed: got %v, expected %v", test.name, errCode, test.expectedErrCode)
+		}
+	}
+}
diff --git a/pkg/gce-cloud-provider/compute/gce_test.go b/pkg/gce-cloud-provider/compute/gce_test.go
new file mode 100644
index 00000000..5bb2aed8
--- /dev/null
+++ b/pkg/gce-cloud-provider/compute/gce_test.go
@@ -0,0 +1,86 @@
+/*
+Copyright 2023 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 gcecloudprovider
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+	"testing"
+
+	"google.golang.org/api/googleapi"
+)
+
+func TestIsGCEError(t *testing.T) {
+	testCases := []struct {
+		name          string
+		inputErr      error
+		reason        string
+		expIsGCEError bool
+	}{
+		{
+			name:          "Not googleapi.Error",
+			inputErr:      errors.New("I am not a googleapi.Error"),
+			reason:        "notFound",
+			expIsGCEError: false,
+		},
+		{
+			name: "googleapi.Error not found error",
+			inputErr: &googleapi.Error{
+				Code: http.StatusNotFound,
+				Errors: []googleapi.ErrorItem{
+					{
+						Reason: "notFound",
+					},
+				},
+				Message: "Not found",
+			},
+			reason:        "notFound",
+			expIsGCEError: true,
+		},
+		{
+			name: "wrapped googleapi.Error",
+			inputErr: fmt.Errorf("encountered not found: %w", &googleapi.Error{
+				Code: http.StatusNotFound,
+				Errors: []googleapi.ErrorItem{
+					{
+						Reason: "notFound",
+					},
+				},
+				Message: "Not found",
+			},
+			),
+			reason:        "notFound",
+			expIsGCEError: true,
+		},
+		{
+			name:          "nil error",
+			inputErr:      nil,
+			reason:        "notFound",
+			expIsGCEError: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Logf("Running test: %v", tc.name)
+		isGCEError := IsGCEError(tc.inputErr, tc.reason)
+		if tc.expIsGCEError != isGCEError {
+			t.Fatalf("Got isGCEError '%t', expected '%t'", isGCEError, tc.expIsGCEError)
+		}
+	}
+}
diff --git a/pkg/gce-pd-csi-driver/controller_test.go b/pkg/gce-pd-csi-driver/controller_test.go
index 254a8369..d49809c7 100644
--- a/pkg/gce-pd-csi-driver/controller_test.go
+++ b/pkg/gce-pd-csi-driver/controller_test.go
@@ -1118,8 +1118,112 @@ func TestCreateVolumeWithVolumeSourceFromSnapshot(t *testing.T) {
 	}
 }
 
+func TestCloningLocationRequirements(t *testing.T) {
+	testSourceVolumeName := "test-volume-source-name"
+	testZonalVolumeSourceID := fmt.Sprintf("projects/%s/zones/%s/disks/%s", project, zone, testSourceVolumeName)
+	testRegionalVolumeSourceID := fmt.Sprintf("projects/%s/regions/%s/disks/%s", project, region, testSourceVolumeName)
+
+	testCases := []struct {
+		name                         string
+		sourceVolumeID               string
+		nilVolumeContentSource       bool
+		reqParameters                map[string]string
+		requestCapacityRange         *csi.CapacityRange
+		replicationType              string
+		expectedLocationRequirements *locationRequirements
+		expectedErr                  bool
+	}{
+		{
+			name:                 "success zonal disk clone of zonal source disk",
+			sourceVolumeID:       testZonalVolumeSourceID,
+			requestCapacityRange: stdCapRange,
+			reqParameters: map[string]string{
+				common.ParameterKeyReplicationType: replicationTypeNone,
+			},
+			replicationType:              replicationTypeNone,
+			expectedLocationRequirements: &locationRequirements{srcVolRegion: region, srcVolZone: zone, srcReplicationType: replicationTypeNone, cloneReplicationType: replicationTypeNone},
+			expectedErr:                  false,
+		},
+		{
+			name:                 "success regional disk clone of regional source disk",
+			sourceVolumeID:       testRegionalVolumeSourceID,
+			requestCapacityRange: stdCapRange,
+			reqParameters: map[string]string{
+				common.ParameterKeyReplicationType: replicationTypeRegionalPD,
+			},
+			replicationType:              replicationTypeRegionalPD,
+			expectedLocationRequirements: &locationRequirements{srcVolRegion: region, srcVolZone: "", srcReplicationType: replicationTypeRegionalPD, cloneReplicationType: replicationTypeRegionalPD},
+			expectedErr:                  false,
+		},
+		{
+			name:                 "success regional disk clone of zonal data source",
+			sourceVolumeID:       testZonalVolumeSourceID,
+			requestCapacityRange: stdCapRange,
+			reqParameters: map[string]string{
+				common.ParameterKeyReplicationType: replicationTypeRegionalPD,
+			},
+			replicationType:              replicationTypeRegionalPD,
+			expectedLocationRequirements: &locationRequirements{srcVolRegion: region, srcVolZone: zone, srcReplicationType: replicationTypeNone, cloneReplicationType: replicationTypeRegionalPD},
+			expectedErr:                  false,
+		},
+		{
+			name:                   "non-cloning CreateVolumeRequest",
+			nilVolumeContentSource: true,
+			requestCapacityRange:   stdCapRange,
+			reqParameters: map[string]string{
+				common.ParameterKeyReplicationType: replicationTypeRegionalPD,
+			},
+			replicationType:              replicationTypeRegionalPD,
+			expectedLocationRequirements: nil,
+			expectedErr:                  false,
+		},
+		{
+			name:                 "failure invalid volumeID",
+			sourceVolumeID:       fmt.Sprintf("projects/%s/disks/%s", project, testSourceVolumeName),
+			requestCapacityRange: stdCapRange,
+			reqParameters: map[string]string{
+				common.ParameterKeyReplicationType: replicationTypeNone,
+			},
+			replicationType:              replicationTypeNone,
+			expectedLocationRequirements: nil,
+			expectedErr:                  true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Logf("test case: %s", tc.name)
+		req := &csi.CreateVolumeRequest{
+			Name:               name,
+			CapacityRange:      tc.requestCapacityRange,
+			VolumeCapabilities: stdVolCaps,
+			Parameters:         tc.reqParameters,
+			VolumeContentSource: &csi.VolumeContentSource{
+				Type: &csi.VolumeContentSource_Volume{
+					Volume: &csi.VolumeContentSource_VolumeSource{
+						VolumeId: tc.sourceVolumeID,
+					},
+				},
+			},
+		}
+		if tc.nilVolumeContentSource {
+			req.VolumeContentSource = nil
+		}
+
+		locationRequirements, err := cloningLocationRequirements(req, tc.replicationType)
+
+		if err != nil != tc.expectedErr {
+			t.Fatalf("Got error %v, expected error %t", err, tc.expectedErr)
+		}
+		input := fmt.Sprintf("cloningLocationRequirements(%v, %s", req, tc.replicationType)
+		if fmt.Sprintf("%v", tc.expectedLocationRequirements) != fmt.Sprintf("%v", locationRequirements) {
+			t.Fatalf("%s returned unexpected diff got: %v, want %v", input, locationRequirements, tc.expectedLocationRequirements)
+		}
+	}
+}
+
 func TestCreateVolumeWithVolumeSourceFromVolume(t *testing.T) {
 	testSourceVolumeName := "test-volume-source-name"
+	testCloneVolumeName := "test-volume-clone"
 	testZonalVolumeSourceID := fmt.Sprintf("projects/%s/zones/%s/disks/%s", project, zone, testSourceVolumeName)
 	testRegionalVolumeSourceID := fmt.Sprintf("projects/%s/regions/%s/disks/%s", project, region, testSourceVolumeName)
 	testSecondZonalVolumeSourceID := fmt.Sprintf("projects/%s/zones/%s/disks/%s", project, "different-zone1", testSourceVolumeName)
@@ -1131,14 +1235,33 @@ func TestCreateVolumeWithVolumeSourceFromVolume(t *testing.T) {
 		common.ParameterKeyType: "test-type", common.ParameterKeyReplicationType: replicationTypeRegionalPD,
 		common.ParameterKeyDiskEncryptionKmsKey: "encryption-key",
 	}
-	topology := &csi.TopologyRequirement{
-		Requisite: []*csi.Topology{
-			{
-				Segments: map[string]string{common.TopologyKeyZone: zone},
-			},
-			{
-				Segments: map[string]string{common.TopologyKeyZone: secondZone},
-			},
+	requisiteTopology := []*csi.Topology{
+		{
+			Segments: map[string]string{common.TopologyKeyZone: zone},
+		},
+		{
+			Segments: map[string]string{common.TopologyKeyZone: secondZone},
+		},
+	}
+
+	requisiteAllRegionZonesTopology := []*csi.Topology{
+		{
+			Segments: map[string]string{common.TopologyKeyZone: "country-region-fakethirdzone"},
+		},
+		{
+			Segments: map[string]string{common.TopologyKeyZone: zone},
+		},
+		{
+			Segments: map[string]string{common.TopologyKeyZone: secondZone},
+		},
+	}
+
+	prefTopology := []*csi.Topology{
+		{
+			Segments: map[string]string{common.TopologyKeyZone: zone},
+		},
+		{
+			Segments: map[string]string{common.TopologyKeyZone: secondZone},
 		},
 	}
 
@@ -1153,39 +1276,326 @@ func TestCreateVolumeWithVolumeSourceFromVolume(t *testing.T) {
 		requestCapacityRange *csi.CapacityRange
 		sourceTopology       *csi.TopologyRequirement
 		requestTopology      *csi.TopologyRequirement
+		expCloneKey          *meta.Key
+		// Accessible topologies validates that the replica zones are valid for regional disk clones.
+		expAccessibleTop []*csi.Topology
 	}{
+
 		{
-			name:                 "success zonal disk clone of zonal source disk",
+			name:                 "success zonal -> zonal cloning, nil topology: immediate binding w/ no allowedTopologies",
 			volumeOnCloud:        true,
 			sourceVolumeID:       testZonalVolumeSourceID,
 			requestCapacityRange: stdCapRange,
 			sourceCapacityRange:  stdCapRange,
 			reqParameters:        zonalParams,
 			sourceReqParameters:  zonalParams,
-			sourceTopology:       topology,
-			requestTopology:      topology,
+			// Source volume will be in the zone that is the first element of preferred topologies (country-region-zone)
+			sourceTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+			requestTopology: nil,
+			expCloneKey:     &meta.Key{Name: testCloneVolumeName, Zone: zone, Region: ""},
+			expAccessibleTop: []*csi.Topology{
+				{
+					Segments: map[string]string{common.TopologyKeyZone: "country-region-zone"},
+				},
+			},
 		},
 		{
-			name:                 "success regional disk clone of regional source disk",
+			name:                 "success zonal -> zonal cloning, req = allowedTopologies, pref = req w/ randomly selected zone as first element: immediate binding w/ allowedTopologies",
 			volumeOnCloud:        true,
-			sourceVolumeID:       testRegionalVolumeSourceID,
+			sourceVolumeID:       testZonalVolumeSourceID,
+			requestCapacityRange: stdCapRange,
+			sourceCapacityRange:  stdCapRange,
+			reqParameters:        zonalParams,
+			sourceReqParameters:  zonalParams,
+			// Source volume will be in the zone that is the first element of preferred topologies (country-region-zone)
+			sourceTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+			requestTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: []*csi.Topology{
+					{
+						Segments: map[string]string{common.TopologyKeyZone: secondZone},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: zone},
+					},
+				},
+			},
+			expCloneKey: &meta.Key{Name: testCloneVolumeName, Zone: zone, Region: ""},
+			expAccessibleTop: []*csi.Topology{
+				{
+					Segments: map[string]string{common.TopologyKeyZone: "country-region-zone"},
+				},
+			},
+		},
+		{
+			name:                 "success zonal -> zonal cloning, req = allowedTopologies, pref = req w/ src zone as first element: delayed binding w/ allowedTopologies",
+			volumeOnCloud:        true,
+			sourceVolumeID:       testZonalVolumeSourceID,
+			requestCapacityRange: stdCapRange,
+			sourceCapacityRange:  stdCapRange,
+			reqParameters:        zonalParams,
+			sourceReqParameters:  zonalParams,
+			// Source volume will be in the zone that is the first element of preferred topologies (country-region-zone)
+			sourceTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+			requestTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+			expCloneKey: &meta.Key{Name: testCloneVolumeName, Zone: zone, Region: ""},
+			expAccessibleTop: []*csi.Topology{
+				{
+					Segments: map[string]string{common.TopologyKeyZone: "country-region-zone"},
+				},
+			},
+		},
+		{
+			name:                 "success zonal -> zonal cloning, req = all zones in region, pref = req w/ src zone as first element: delayed binding without allowedTopologies",
+			volumeOnCloud:        true,
+			sourceVolumeID:       testZonalVolumeSourceID,
+			requestCapacityRange: stdCapRange,
+			sourceCapacityRange:  stdCapRange,
+			reqParameters:        zonalParams,
+			sourceReqParameters:  zonalParams,
+			// Source volume will be in the zone that is the first element of preferred topologies (country-region-zone)
+			sourceTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+			requestTopology: &csi.TopologyRequirement{
+				Requisite: requisiteAllRegionZonesTopology,
+				Preferred: prefTopology,
+			},
+			expCloneKey: &meta.Key{Name: testCloneVolumeName, Zone: zone, Region: ""},
+			expAccessibleTop: []*csi.Topology{
+				{
+					Segments: map[string]string{common.TopologyKeyZone: "country-region-zone"},
+				},
+			},
+		},
+		{
+			name:                 "success zonal -> regional cloning, nil topology: immediate binding w/ no allowedTopologies",
+			volumeOnCloud:        true,
+			sourceVolumeID:       testZonalVolumeSourceID,
 			requestCapacityRange: stdCapRange,
 			sourceCapacityRange:  stdCapRange,
 			reqParameters:        regionalParams,
-			sourceReqParameters:  regionalParams,
-			sourceTopology:       topology,
-			requestTopology:      topology,
+			sourceReqParameters:  zonalParams,
+			sourceTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+			requestTopology: nil,
+			expCloneKey:     &meta.Key{Name: testCloneVolumeName, Zone: "", Region: "country-region"},
+			expAccessibleTop: []*csi.Topology{
+				{
+					Segments: map[string]string{common.TopologyKeyZone: "country-region-zone"},
+				},
+				{
+					Segments: map[string]string{common.TopologyKeyZone: "country-region-fakesecondzone"},
+				},
+			},
 		},
 		{
-			name:                 "success regional disk clone of zonal data source",
+			name:                 "success zonal -> regional cloning, req = allowedTopologies, pref = req w/ randomly selected zone as first element: immediate binding w/ allowedTopologies",
+			volumeOnCloud:        true,
+			sourceVolumeID:       testZonalVolumeSourceID,
+			requestCapacityRange: stdCapRange,
+			sourceCapacityRange:  stdCapRange,
+			reqParameters:        regionalParams,
+			sourceReqParameters:  zonalParams,
+			sourceTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+			requestTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: []*csi.Topology{
+					{
+						Segments: map[string]string{common.TopologyKeyZone: secondZone},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: zone},
+					},
+				},
+			},
+			expCloneKey: &meta.Key{Name: testCloneVolumeName, Zone: "", Region: "country-region"},
+			expAccessibleTop: []*csi.Topology{
+				{
+					Segments: map[string]string{common.TopologyKeyZone: "country-region-zone"},
+				},
+				{
+					Segments: map[string]string{common.TopologyKeyZone: "country-region-fakesecondzone"},
+				},
+			},
+		},
+		{
+			name:                 "success zonal -> regional cloning, req = allowedTopologies, pref = req w/ src zone as first element: delayed binding w/ allowedTopologies",
+			volumeOnCloud:        true,
+			sourceVolumeID:       testZonalVolumeSourceID,
+			requestCapacityRange: stdCapRange,
+			sourceCapacityRange:  stdCapRange,
+			reqParameters:        regionalParams,
+			sourceReqParameters:  zonalParams,
+			sourceTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+			requestTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+			expCloneKey: &meta.Key{Name: testCloneVolumeName, Zone: "", Region: "country-region"},
+			expAccessibleTop: []*csi.Topology{
+				{
+					Segments: map[string]string{common.TopologyKeyZone: "country-region-zone"},
+				},
+				{
+					Segments: map[string]string{common.TopologyKeyZone: "country-region-fakesecondzone"},
+				},
+			},
+		},
+		{
+			name:                 "success zonal -> regional cloning, req = all zones in region, pref = req w/ src zone as first element: delayed binding without allowedTopologies",
 			volumeOnCloud:        true,
 			sourceVolumeID:       testZonalVolumeSourceID,
 			requestCapacityRange: stdCapRange,
 			sourceCapacityRange:  stdCapRange,
 			reqParameters:        regionalParams,
 			sourceReqParameters:  zonalParams,
-			sourceTopology:       topology,
-			requestTopology:      topology,
+			sourceTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+			requestTopology: &csi.TopologyRequirement{
+				Requisite: requisiteAllRegionZonesTopology,
+				Preferred: prefTopology,
+			},
+			expCloneKey: &meta.Key{Name: testCloneVolumeName, Zone: "", Region: "country-region"},
+			expAccessibleTop: []*csi.Topology{
+				{
+					Segments: map[string]string{common.TopologyKeyZone: "country-region-zone"},
+				},
+				{
+					Segments: map[string]string{common.TopologyKeyZone: "country-region-fakesecondzone"},
+				},
+			},
+		},
+		{
+			name:                 "success regional -> regional cloning, nil topology: immediate binding w/ no allowedTopologies",
+			volumeOnCloud:        true,
+			sourceVolumeID:       testRegionalVolumeSourceID,
+			requestCapacityRange: stdCapRange,
+			sourceCapacityRange:  stdCapRange,
+			reqParameters:        regionalParams,
+			sourceReqParameters:  regionalParams,
+			sourceTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+			requestTopology: nil,
+			expCloneKey:     &meta.Key{Name: testCloneVolumeName, Zone: "", Region: "country-region"},
+			expAccessibleTop: []*csi.Topology{
+				{
+					Segments: map[string]string{common.TopologyKeyZone: "country-region-zone"},
+				},
+				{
+					Segments: map[string]string{common.TopologyKeyZone: "country-region-fakesecondzone"},
+				},
+			},
+		},
+		{
+			name:                 "success regional -> regional cloning, req = allowedTopologies, pref = req w/ randomly selected zone as first element: immediate binding w/ allowedTopologies",
+			volumeOnCloud:        true,
+			sourceVolumeID:       testRegionalVolumeSourceID,
+			requestCapacityRange: stdCapRange,
+			sourceCapacityRange:  stdCapRange,
+			reqParameters:        regionalParams,
+			sourceReqParameters:  regionalParams,
+			sourceTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+			requestTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: []*csi.Topology{
+					{
+						Segments: map[string]string{common.TopologyKeyZone: secondZone},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: zone},
+					},
+				},
+			},
+			expCloneKey: &meta.Key{Name: testCloneVolumeName, Zone: "", Region: "country-region"},
+			expAccessibleTop: []*csi.Topology{
+				{
+					Segments: map[string]string{common.TopologyKeyZone: "country-region-zone"},
+				},
+				{
+					Segments: map[string]string{common.TopologyKeyZone: "country-region-fakesecondzone"},
+				},
+			},
+		},
+		{
+			name:                 "success regional -> regional cloning, req = allowedTopologies, pref = req w/ src zone as first element: delayed binding w/ allowedTopologies",
+			volumeOnCloud:        true,
+			sourceVolumeID:       testRegionalVolumeSourceID,
+			requestCapacityRange: stdCapRange,
+			sourceCapacityRange:  stdCapRange,
+			reqParameters:        regionalParams,
+			sourceReqParameters:  regionalParams,
+			sourceTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+			requestTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+			expCloneKey: &meta.Key{Name: testCloneVolumeName, Zone: "", Region: "country-region"},
+			expAccessibleTop: []*csi.Topology{
+				{
+					Segments: map[string]string{common.TopologyKeyZone: "country-region-zone"},
+				},
+				{
+					Segments: map[string]string{common.TopologyKeyZone: "country-region-fakesecondzone"},
+				},
+			},
+		},
+		{
+			name:                 "success regional -> regional cloning, req = all zones in region, pref = req w/ src zone as first element: delayed binding without allowedTopologies",
+			volumeOnCloud:        true,
+			sourceVolumeID:       testRegionalVolumeSourceID,
+			requestCapacityRange: stdCapRange,
+			sourceCapacityRange:  stdCapRange,
+			reqParameters:        regionalParams,
+			sourceReqParameters:  regionalParams,
+			sourceTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+			requestTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+			expCloneKey: &meta.Key{Name: testCloneVolumeName, Zone: "", Region: "country-region"},
+			expAccessibleTop: []*csi.Topology{
+				{
+					Segments: map[string]string{common.TopologyKeyZone: "country-region-zone"},
+				},
+				{
+					Segments: map[string]string{common.TopologyKeyZone: "country-region-fakesecondzone"},
+				},
+			},
 		},
 		{
 			name:                 "fail regional disk clone with no matching replica zone of zonal data source",
@@ -1196,7 +1606,10 @@ func TestCreateVolumeWithVolumeSourceFromVolume(t *testing.T) {
 			sourceCapacityRange:  stdCapRange,
 			reqParameters:        regionalParams,
 			sourceReqParameters:  zonalParams,
-			sourceTopology:       topology,
+			sourceTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
 			requestTopology: &csi.TopologyRequirement{
 				Requisite: []*csi.Topology{
 					{
@@ -1219,8 +1632,14 @@ func TestCreateVolumeWithVolumeSourceFromVolume(t *testing.T) {
 			sourceReqParameters: map[string]string{
 				common.ParameterKeyType: "different-type",
 			},
-			sourceTopology:  topology,
-			requestTopology: topology,
+			sourceTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+			requestTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
 		},
 		{
 			name:                 "fail zonal disk clone with different DiskEncryptionKMSKey",
@@ -1234,8 +1653,14 @@ func TestCreateVolumeWithVolumeSourceFromVolume(t *testing.T) {
 				common.ParameterKeyType: "test-type", common.ParameterKeyReplicationType: replicationTypeNone,
 				common.ParameterKeyDiskEncryptionKmsKey: "different-encryption-key",
 			},
-			sourceTopology:  topology,
-			requestTopology: topology,
+			sourceTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+			requestTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
 		},
 		{
 			name:                 "fail zonal disk clone with different zone",
@@ -1256,7 +1681,10 @@ func TestCreateVolumeWithVolumeSourceFromVolume(t *testing.T) {
 					},
 				},
 			},
-			requestTopology: topology,
+			requestTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
 		},
 		{
 			name:                 "fail zonal disk clone of regional data source",
@@ -1267,8 +1695,14 @@ func TestCreateVolumeWithVolumeSourceFromVolume(t *testing.T) {
 			sourceCapacityRange:  stdCapRange,
 			reqParameters:        zonalParams,
 			sourceReqParameters:  regionalParams,
-			sourceTopology:       topology,
-			requestTopology:      topology,
+			sourceTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+			requestTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
 		},
 
 		{
@@ -1280,7 +1714,10 @@ func TestCreateVolumeWithVolumeSourceFromVolume(t *testing.T) {
 			sourceCapacityRange:  stdCapRange,
 			reqParameters:        stdParams,
 			sourceReqParameters:  stdParams,
-			requestTopology:      topology,
+			requestTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
 		},
 		{
 			name:                 "fail invalid source disk volume id format",
@@ -1291,7 +1728,10 @@ func TestCreateVolumeWithVolumeSourceFromVolume(t *testing.T) {
 			sourceCapacityRange:  stdCapRange,
 			reqParameters:        stdParams,
 			sourceReqParameters:  stdParams,
-			requestTopology:      topology,
+			requestTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
 		},
 		{
 			name:                 "fail zonal disk clone with smaller disk capacity",
@@ -1304,16 +1744,23 @@ func TestCreateVolumeWithVolumeSourceFromVolume(t *testing.T) {
 			},
 			reqParameters:       zonalParams,
 			sourceReqParameters: zonalParams,
-			sourceTopology:      topology,
-			requestTopology:     topology,
-		}}
+			sourceTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+			requestTopology: &csi.TopologyRequirement{
+				Requisite: requisiteTopology,
+				Preferred: prefTopology,
+			},
+		},
+	}
 
 	for _, tc := range testCases {
 		t.Logf("test case: %s", tc.name)
 		gceDriver := initGCEDriver(t, nil)
 
 		req := &csi.CreateVolumeRequest{
-			Name:               name,
+			Name:               testCloneVolumeName,
 			CapacityRange:      tc.requestCapacityRange,
 			VolumeCapabilities: stdVolCaps,
 			Parameters:         tc.reqParameters,
@@ -1348,7 +1795,6 @@ func TestCreateVolumeWithVolumeSourceFromVolume(t *testing.T) {
 		}
 
 		resp, err := gceDriver.cs.CreateVolume(context.Background(), req)
-		t.Logf("response: %v err: %v", resp, err)
 		if err != nil {
 			serverError, ok := status.FromError(err)
 			if !ok {
@@ -1364,14 +1810,39 @@ func TestCreateVolumeWithVolumeSourceFromVolume(t *testing.T) {
 		}
 
 		// Make sure the response has the source volume.
-		sourceVolume := resp.GetVolume()
-		if sourceVolume.ContentSource == nil || sourceVolume.ContentSource.Type == nil ||
-			sourceVolume.ContentSource.GetVolume() == nil || sourceVolume.ContentSource.GetVolume().VolumeId == "" {
+		respVolume := resp.GetVolume()
+		if respVolume.ContentSource == nil || respVolume.ContentSource.Type == nil ||
+			respVolume.ContentSource.GetVolume() == nil || respVolume.ContentSource.GetVolume().VolumeId == "" {
 			t.Fatalf("Expected volume content source to have volume ID, got none")
 		}
+		// Validate that the cloned volume is in the region/zone that we expect
+		cloneVolID := respVolume.VolumeId
+		_, cloneVolKey, err := common.VolumeIDToKey(cloneVolID)
+		if err != nil {
+			t.Fatalf("failed to get key from volume id %q: %v", cloneVolID, err)
+		}
+		if cloneVolKey.String() != tc.expCloneKey.String() {
+			t.Fatalf("got clone volume key: %q, expected clone volume key: %q", cloneVolKey.String(), tc.expCloneKey.String())
+		}
+		if !accessibleTopologiesEqual(respVolume.AccessibleTopology, tc.expAccessibleTop) {
+			t.Fatalf("got accessible topology: %q, expected accessible topology: %q", fmt.Sprintf("%+v", respVolume.AccessibleTopology), fmt.Sprintf("%+v", tc.expAccessibleTop))
+
+		}
 	}
 }
 
+func sortTopologies(in []*csi.Topology) {
+	sort.Slice(in, func(i, j int) bool {
+		return in[i].Segments[common.TopologyKeyZone] < in[j].Segments[common.TopologyKeyZone]
+	})
+}
+
+func accessibleTopologiesEqual(got []*csi.Topology, expected []*csi.Topology) bool {
+	sortTopologies(got)
+	sortTopologies(expected)
+	return fmt.Sprintf("%+v", got) == fmt.Sprintf("%+v", expected)
+}
+
 func TestCreateVolumeRandomRequisiteTopology(t *testing.T) {
 	req := &csi.CreateVolumeRequest{
 		Name:               "test-name",
@@ -1802,10 +2273,93 @@ func TestGetZonesFromTopology(t *testing.T) {
 	}
 }
 
+func TestPickZonesInRegion(t *testing.T) {
+	testCases := []struct {
+		name     string
+		region   string
+		zones    []string
+		expZones []string
+	}{
+		{
+			name:     "all zones in region",
+			region:   "us-central1",
+			zones:    []string{"us-central1-a", "us-central1-b", "us-central1-c"},
+			expZones: []string{"us-central1-a", "us-central1-b", "us-central1-c"},
+		},
+		{
+			name:     "removes zones not in region",
+			region:   "us-central1",
+			zones:    []string{"us-central1-a", "us-central1-b", "us-central1-c", "us-east1-a, us-west1-a"},
+			expZones: []string{"us-central1-a", "us-central1-b", "us-central1-c"},
+		},
+		{
+			name:     "region not in zones",
+			region:   "us-west1",
+			zones:    []string{"us-central1-a", "us-central1-b", "us-central1-c"},
+			expZones: []string{},
+		},
+		{
+			name:     "empty zones",
+			region:   "us-central1",
+			zones:    []string{},
+			expZones: []string{},
+		},
+	}
+	for _, tc := range testCases {
+		t.Logf("test case: %s", tc.name)
+		gotZones := pickZonesInRegion(tc.region, tc.zones)
+		if !sets.NewString(gotZones...).Equal(sets.NewString(tc.expZones...)) {
+			t.Errorf("Got zones: %v, expected: %v", gotZones, tc.expZones)
+		}
+	}
+}
+
+func TestPrependZone(t *testing.T) {
+	testCases := []struct {
+		name     string
+		zone     string
+		zones    []string
+		expZones []string
+	}{
+		{
+			name:     "zone already at index 0",
+			zone:     "us-central1-a",
+			zones:    []string{"us-central1-a", "us-central1-b", "us-central1-c"},
+			expZones: []string{"us-central1-a", "us-central1-b", "us-central1-c"},
+		},
+		{
+			name:     "zone at index 1",
+			zone:     "us-central1-b",
+			zones:    []string{"us-central1-a", "us-central1-b", "us-central1-c"},
+			expZones: []string{"us-central1-b", "us-central1-a", "us-central1-c"},
+		},
+		{
+			name:     "zone not in zones",
+			zone:     "us-central1-f",
+			zones:    []string{"us-central1-a", "us-central1-b", "us-central1-c"},
+			expZones: []string{"us-central1-f", "us-central1-a", "us-central1-b", "us-central1-c"},
+		},
+		{
+			name:     "empty zones",
+			zone:     "us-central1-a",
+			zones:    []string{},
+			expZones: []string{"us-central1-a"},
+		},
+	}
+	for _, tc := range testCases {
+		t.Logf("test case: %s", tc.name)
+		gotZones := prependZone(tc.zone, tc.zones)
+		if !zonesEqual(gotZones, tc.expZones) {
+			t.Errorf("Got zones: %v, expected: %v", gotZones, tc.expZones)
+		}
+	}
+}
+
 func TestPickZonesFromTopology(t *testing.T) {
 	testCases := []struct {
 		name     string
 		top      *csi.TopologyRequirement
+		locReq   *locationRequirements
 		numZones int
 		expZones []string
 		expErr   bool
@@ -1839,6 +2393,36 @@ func TestPickZonesFromTopology(t *testing.T) {
 			numZones: 2,
 			expZones: []string{"topology-zone2", "topology-zone3"},
 		},
+		{
+			name: "success: preferred, locationRequirements[region:us-central1, zone:us-central1-a, srcReplicationType:none, cloneReplicationType:none]",
+			top: &csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-c"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-b"},
+					},
+				},
+				Preferred: []*csi.Topology{
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-b"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-c"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+					},
+				},
+			},
+			locReq:   &locationRequirements{srcVolRegion: "us-central1", srcVolZone: "us-central1-a", srcReplicationType: replicationTypeNone, cloneReplicationType: replicationTypeNone},
+			numZones: 1,
+			expZones: []string{"us-central1-a"},
+		},
 		{
 			name: "success: preferred and requisite",
 			top: &csi.TopologyRequirement{
@@ -1874,6 +2458,72 @@ func TestPickZonesFromTopology(t *testing.T) {
 			numZones: 5,
 			expZones: []string{"topology-zone2", "topology-zone3", "topology-zone1", "topology-zone5", "topology-zone6"},
 		},
+		{
+			name: "success: preferred and requisite, locationRequirements[region:us-central1, zone:us-central1-a, srcReplicationType:regional-pd, cloneReplicationType:regional-pd]",
+			top: &csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-c"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-d"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-f"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-west1-a"},
+					},
+				},
+				Preferred: []*csi.Topology{
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-b"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-east1-a"},
+					},
+				},
+			},
+			locReq:   &locationRequirements{srcVolRegion: "us-central1", srcVolZone: "us-central1-a", srcReplicationType: replicationTypeRegionalPD, cloneReplicationType: replicationTypeRegionalPD},
+			numZones: 5,
+			expZones: []string{"us-central1-b", "us-central1-c", "us-central1-a", "us-central1-d", "us-central1-f"},
+		},
+		{
+			name: "success: preferred and requisite, locationRequirements[region:us-central1, zone:us-central1-a, srcReplicationType:none, cloneReplicationType:regional-pd]",
+			top: &csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-c"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-d"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-f"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-west1-a"},
+					},
+				},
+				Preferred: []*csi.Topology{
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-b"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-east1-a"},
+					},
+				},
+			},
+			locReq:   &locationRequirements{srcVolRegion: "us-central1", srcVolZone: "us-central1-a", srcReplicationType: replicationTypeNone, cloneReplicationType: replicationTypeRegionalPD},
+			numZones: 5,
+			expZones: []string{"us-central1-a", "us-central1-b", "us-central1-c", "us-central1-d", "us-central1-f"},
+		},
 		{
 			name: "fail: not enough topologies",
 			top: &csi.TopologyRequirement{
@@ -1903,6 +2553,114 @@ func TestPickZonesFromTopology(t *testing.T) {
 			numZones: 4,
 			expErr:   true,
 		},
+		{
+			name: "fail: no topologies that match locationRequirment, locationRequirements[region:us-east1, zone:us-east1-a, replicationType:none]",
+			top: &csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-c"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-b"},
+					},
+				},
+				Preferred: []*csi.Topology{
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-b"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-c"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+					},
+				},
+			},
+			locReq:   &locationRequirements{srcVolRegion: "us-east1", srcVolZone: "us-east1-a", cloneReplicationType: replicationTypeNone},
+			numZones: 1,
+			expErr:   true,
+		},
+		{
+			name: "fail: no topologies that match locationRequirment, locationRequirements[region:us-east1, zone:us-east1-a, replicationType:regional-pd]",
+			top: &csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-c"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-b"},
+					},
+				},
+				Preferred: []*csi.Topology{
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-b"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-c"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+					},
+				},
+			},
+			locReq:   &locationRequirements{srcVolRegion: "us-east1", srcVolZone: "us-east1-a", cloneReplicationType: replicationTypeRegionalPD},
+			numZones: 2,
+			expErr:   true,
+		},
+		{
+			name: "fail: not enough topologies, locationRequirements[region:us-central1, zone:us-central1-a, replicationType:regional-pd]",
+			top: &csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-c"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-b"},
+					},
+				},
+				Preferred: []*csi.Topology{
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-b"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-c"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+					},
+				},
+			},
+			locReq:   &locationRequirements{srcVolRegion: "us-central1", srcVolZone: "us-central1-a", cloneReplicationType: replicationTypeRegionalPD},
+			numZones: 4,
+			expErr:   true,
+		},
+		{
+			name: "success: only requisite, locationRequirements[region:us-central1, zone:us-central1-a, replicationType:regional-pd",
+			top: &csi.TopologyRequirement{
+				Requisite: []*csi.Topology{
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-c"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-a"},
+					},
+					{
+						Segments: map[string]string{common.TopologyKeyZone: "us-central1-b"},
+					},
+				},
+			},
+			numZones: 3,
+			expZones: []string{"us-central1-b", "us-central1-c", "us-central1-a"},
+		},
 		{
 			name: "success: only requisite",
 			top: &csi.TopologyRequirement{
@@ -1924,12 +2682,12 @@ func TestPickZonesFromTopology(t *testing.T) {
 	}
 	for _, tc := range testCases {
 		t.Logf("test case: %s", tc.name)
-		gotZones, err := pickZonesFromTopology(tc.top, tc.numZones)
+		gotZones, err := pickZonesFromTopology(tc.top, tc.numZones, tc.locReq)
 		if err != nil && !tc.expErr {
-			t.Errorf("Did not expect error but got: %v", err)
+			t.Errorf("got error: %v, but did not expect error", err)
 		}
 		if err == nil && tc.expErr {
-			t.Errorf("Expected error but got none")
+			t.Errorf("got no error, but expected error")
 		}
 		if !sets.NewString(gotZones...).Equal(sets.NewString(tc.expZones...)) {
 			t.Errorf("Expected zones: %v, but got: %v", tc.expZones, gotZones)
@@ -1937,6 +2695,18 @@ func TestPickZonesFromTopology(t *testing.T) {
 	}
 }
 
+func zonesEqual(gotZones, expectedZones []string) bool {
+	if len(gotZones) != len(expectedZones) {
+		return false
+	}
+	for i := 0; i < len(gotZones); i++ {
+		if gotZones[i] != expectedZones[i] {
+			return false
+		}
+	}
+	return true
+}
+
 func TestPickRandAndConsecutive(t *testing.T) {
 	rand.Seed(time.Now().UnixNano())
 	testCases := []struct {
diff --git a/pkg/gce-pd-csi-driver/utils_test.go b/pkg/gce-pd-csi-driver/utils_test.go
index ab92e45e..2a0fd5f4 100644
--- a/pkg/gce-pd-csi-driver/utils_test.go
+++ b/pkg/gce-pd-csi-driver/utils_test.go
@@ -18,13 +18,9 @@ limitations under the License.
 package gceGCEDriver
 
 import (
-	"errors"
-	"net/http"
 	"testing"
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
-	"google.golang.org/api/googleapi"
-	"google.golang.org/grpc/codes"
 )
 
 var (
@@ -295,37 +291,3 @@ func TestGetReadOnlyFromCapabilities(t *testing.T) {
 		}
 	}
 }
-
-func TestCodeForError(t *testing.T) {
-	internalErrorCode := codes.Internal
-	userErrorCode := codes.InvalidArgument
-	testCases := []struct {
-		name     string
-		inputErr error
-		expCode  *codes.Code
-	}{
-		{
-			name:     "Not googleapi.Error",
-			inputErr: errors.New("I am not a googleapi.Error"),
-			expCode:  &internalErrorCode,
-		},
-		{
-			name:     "User error",
-			inputErr: &googleapi.Error{Code: http.StatusBadRequest, Message: "User error with bad request"},
-			expCode:  &userErrorCode,
-		},
-		{
-			name:     "googleapi.Error but not a user error",
-			inputErr: &googleapi.Error{Code: http.StatusInternalServerError, Message: "Internal error"},
-			expCode:  &internalErrorCode,
-		},
-	}
-
-	for _, tc := range testCases {
-		t.Logf("Running test: %v", tc.name)
-		actualCode := *CodeForError(tc.inputErr)
-		if *tc.expCode != actualCode {
-			t.Fatalf("Expected error code '%v' but got '%v'", tc.expCode, actualCode)
-		}
-	}
-}
diff --git a/test/e2e/tests/single_zone_e2e_test.go b/test/e2e/tests/single_zone_e2e_test.go
index 7c8d7e05..96075799 100644
--- a/test/e2e/tests/single_zone_e2e_test.go
+++ b/test/e2e/tests/single_zone_e2e_test.go
@@ -359,8 +359,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 		Expect(cloudDisk.Name).To(Equal(volName))
 		Expect(len(cloudDisk.ReplicaZones)).To(Equal(2))
 		for _, replicaZone := range cloudDisk.ReplicaZones {
-			tokens := strings.Split(replicaZone, "/")
-			actualZone := tokens[len(tokens)-1]
+			actualZone := zoneFromURL(replicaZone)
 			gotRegion, err := common.GetRegionFromZones([]string{actualZone})
 			Expect(err).To(BeNil(), "failed to get region from actual zone %v", actualZone)
 			Expect(gotRegion).To(Equal(region), "Got region from replica zone that did not match supplied region")
@@ -677,8 +676,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 		Expect(cloudDisk.Name).To(Equal(volName))
 		Expect(len(cloudDisk.ReplicaZones)).To(Equal(2))
 		for _, replicaZone := range cloudDisk.ReplicaZones {
-			tokens := strings.Split(replicaZone, "/")
-			actualZone := tokens[len(tokens)-1]
+			actualZone := zoneFromURL(replicaZone)
 			gotRegion, err := common.GetRegionFromZones([]string{actualZone})
 			Expect(err).To(BeNil(), "failed to get region from actual zone %v", actualZone)
 			Expect(gotRegion).To(Equal(region), "Got region from replica zone that did not match supplied region")
@@ -1071,6 +1069,10 @@ var _ = Describe("GCE PD CSI Driver", func() {
 		Expect(cloudDisk.Status).To(Equal(readyState))
 		Expect(cloudDisk.SizeGb).To(Equal(defaultSizeGb))
 		Expect(cloudDisk.Name).To(Equal(volName))
+		// Validate the the clone disk zone matches the source disk zone.
+		_, srcKey, err := common.VolumeIDToKey(srcVolID)
+		Expect(err).To(BeNil(), "Could not get source volume key from id")
+		Expect(zoneFromURL(cloudDisk.Zone)).To(Equal(srcKey.Zone))
 		defer func() {
 			// Delete Disk
 			controllerClient.DeleteVolume(volID)
@@ -1122,9 +1124,77 @@ var _ = Describe("GCE PD CSI Driver", func() {
 		Expect(cloudDisk.SizeGb).To(Equal(defaultRepdSizeGb))
 		Expect(cloudDisk.Name).To(Equal(volName))
 		Expect(len(cloudDisk.ReplicaZones)).To(Equal(2))
+		replicaZonesCompatible := false
+		_, srcKey, err := common.VolumeIDToKey(srcVolID)
+		Expect(err).To(BeNil(), "Could not get source volume key from id")
 		for _, replicaZone := range cloudDisk.ReplicaZones {
-			tokens := strings.Split(replicaZone, "/")
-			actualZone := tokens[len(tokens)-1]
+			actualZone := zoneFromURL(replicaZone)
+			if actualZone == srcKey.Zone {
+				replicaZonesCompatible = true
+			}
+			gotRegion, err := common.GetRegionFromZones([]string{actualZone})
+			Expect(err).To(BeNil(), "failed to get region from actual zone %v", actualZone)
+			Expect(gotRegion).To(Equal(region), "Got region from replica zone that did not match supplied region")
+		}
+		// Validate that one of the replicaZones of the clone matches the zone of the source disk.
+		Expect(replicaZonesCompatible).To(Equal(true))
+		defer func() {
+			// Delete Disk
+			controllerClient.DeleteVolume(volID)
+			Expect(err).To(BeNil(), "DeleteVolume failed")
+
+			// Validate Disk Deleted
+			_, err = computeService.RegionDisks.Get(p, region, volName).Do()
+			Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found")
+		}()
+	})
+
+	It("Should successfully create RePD from a RePD VolumeContentSource", func() {
+		Expect(testContexts).ToNot(BeEmpty())
+		testContext := getRandomTestContext()
+
+		controllerInstance := testContext.Instance
+		controllerClient := testContext.Client
+
+		p, z, _ := controllerInstance.GetIdentity()
+
+		region, err := common.GetRegionFromZones([]string{z})
+		Expect(err).To(BeNil(), "Failed to get region from zones")
+
+		// Create Source Disk
+		srcVolName := testNamePrefix + string(uuid.NewUUID())
+		srcVolID, err := controllerClient.CreateVolume(srcVolName, map[string]string{
+			common.ParameterKeyReplicationType: "regional-pd",
+		}, defaultRepdSizeGb, nil, nil)
+		// Create Disk
+		volName := testNamePrefix + string(uuid.NewUUID())
+		volID, err := controllerClient.CreateVolume(volName, map[string]string{
+			common.ParameterKeyReplicationType: "regional-pd",
+		}, defaultRepdSizeGb, nil,
+			&csi.VolumeContentSource{
+				Type: &csi.VolumeContentSource_Volume{
+					Volume: &csi.VolumeContentSource_VolumeSource{
+						VolumeId: srcVolID,
+					},
+				},
+			})
+
+		Expect(err).To(BeNil(), "CreateVolume failed with error: %v", err)
+
+		// Validate Disk Created
+		cloudDisk, err := computeService.RegionDisks.Get(p, region, volName).Do()
+		Expect(err).To(BeNil(), "Could not get disk from cloud directly")
+		Expect(cloudDisk.Type).To(ContainSubstring(standardDiskType))
+		Expect(cloudDisk.Status).To(Equal(readyState))
+		Expect(cloudDisk.SizeGb).To(Equal(defaultRepdSizeGb))
+		Expect(cloudDisk.Name).To(Equal(volName))
+		Expect(len(cloudDisk.ReplicaZones)).To(Equal(2))
+		// Validate that the replicaZones of the clone match the replicaZones of the source disk.
+		srcCloudDisk, err := computeService.RegionDisks.Get(p, region, srcVolName).Do()
+		Expect(err).To(BeNil(), "Could not get source disk from cloud directly")
+		Expect(srcCloudDisk.ReplicaZones).To(Equal(cloudDisk.ReplicaZones))
+		for _, replicaZone := range cloudDisk.ReplicaZones {
+			actualZone := zoneFromURL(replicaZone)
 			gotRegion, err := common.GetRegionFromZones([]string{actualZone})
 			Expect(err).To(BeNil(), "failed to get region from actual zone %v", actualZone)
 			Expect(gotRegion).To(Equal(region), "Got region from replica zone that did not match supplied region")
@@ -1263,6 +1333,16 @@ func cleanSelfLink(selfLink string) string {
 	return r.ReplaceAllString(selfLink, "")
 }
 
+// Returns the zone from the URL with the format https://compute.googleapis.com/compute/v1/projects/{project}/zones/{zone}.
+// Returns the empty string if the zone cannot be abstracted from the URL.
+func zoneFromURL(url string) string {
+	tokens := strings.Split(url, "/")
+	if len(tokens) == 0 {
+		return ""
+	}
+	return tokens[len(tokens)-1]
+}
+
 func setupKeyRing(ctx context.Context, parentName string, keyRingId string) (*kmspb.CryptoKey, []string) {
 	// Create KeyRing
 	ringReq := &kmspb.CreateKeyRingRequest{
diff --git a/test/remote/instance.go b/test/remote/instance.go
index c716dcef..8aa5b2fd 100644
--- a/test/remote/instance.go
+++ b/test/remote/instance.go
@@ -415,8 +415,8 @@ func generateMetadataWithPublicKey(pubKeyFile string) (*compute.Metadata, error)
 // isGCEError returns true if given error is a googleapi.Error with given
 // reason (e.g. "resourceInUseByAnotherResource")
 func isGCEError(err error, reason string) bool {
-	apiErr, ok := err.(*googleapi.Error)
-	if !ok {
+	var apiErr *googleapi.Error
+	if !errors.As(err, &apiErr) {
 		return false
 	}