Add support for a multi-zone volumeHandle

Author: pwschuurman

pkg/common/constants.go

@@ -24,4 +24,12 @@ const (
 	VolumeAttributePartition = "partition"
 
 	UnspecifiedValue = "UNSPECIFIED"
+
+	// Keyword indicating a 'multi-zone' volumeHandle. Replaces "zones" in the volumeHandle:
+	// eg: projects/{project}/zones/multi-zone/disks/{name} vs.
+	// projects/{project}/zones/{zone}/disks/{name}
+	MultiZoneValue = "multi-zone"
+
+	// Label that is set on a disk when it is used by a 'multi-zone' VolumeHandle
+	MultiZoneLabel = "goog-gke-multi-zone"
 )

pkg/common/utils.go

@@ -478,3 +478,15 @@ func UnorderedSlicesEqual(slice1 []string, slice2 []string) bool {
 	}
 	return true
 }
+
+func VolumeIdAsMultiZone(volumeId string) (string, error) {
+	splitId := strings.Split(volumeId, "/")
+	if len(splitId) != volIDTotalElements {
+		return "", fmt.Errorf("failed to get id components. Expected projects/{project}/zones/{zone}/disks/{name}. Got: %s", volumeId)
+	}
+	if splitId[volIDToplogyKey] != "zones" {
+		return "", fmt.Errorf("expected id to be zonal. Got: %s", volumeId)
+	}
+	splitId[volIDToplogyValue] = MultiZoneValue
+	return strings.Join(splitId, "/"), nil
+}

pkg/gce-cloud-provider/compute/fake-gce.go

@@ -92,6 +92,10 @@ func (cloud *FakeCloudProvider) GetDefaultZone() string {
 	return cloud.zone
 }
 
+func (cloud *FakeCloudProvider) RepairUnderspecifiedVolumeKeyForZone(ctx context.Context, project string, volumeKey *meta.Key, instanceZone string) (string, *meta.Key, error) {
+	return cloud.RepairUnderspecifiedVolumeKey(ctx, project, volumeKey)
+}
+
 func (cloud *FakeCloudProvider) RepairUnderspecifiedVolumeKey(ctx context.Context, project string, volumeKey *meta.Key) (string, *meta.Key, error) {
 	if project == common.UnspecifiedValue {
 		project = cloud.project

pkg/gce-cloud-provider/compute/gce-compute.go

@@ -91,6 +91,7 @@ type GCECompute interface {
 	GetDefaultZone() string
 	// Disk Methods
 	GetDisk(ctx context.Context, project string, volumeKey *meta.Key, gceAPIVersion GCEAPIVersion) (*CloudDisk, error)
+	RepairUnderspecifiedVolumeKeyForZone(ctx context.Context, project string, volumeKey *meta.Key, instanceZone string) (string, *meta.Key, error)
 	RepairUnderspecifiedVolumeKey(ctx context.Context, project string, volumeKey *meta.Key) (string, *meta.Key, error)
 	ValidateExistingDisk(ctx context.Context, disk *CloudDisk, params common.DiskParameters, reqBytes, limBytes int64, multiWriter bool) error
 	InsertDisk(ctx context.Context, project string, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool) error
@@ -171,6 +172,17 @@ func (cloud *CloudProvider) ListDisks(ctx context.Context) ([]*computev1.Disk, s
 	return items, "", nil
 }
 
+// RepairUnderspecifiedVolumeKey will query the cloud provider and check each zone for the disk specified
+// by the volume key and return a volume key with a correct zone
+func (cloud *CloudProvider) RepairUnderspecifiedVolumeKeyForZone(ctx context.Context, project string, volumeKey *meta.Key, instanceZone string) (string, *meta.Key, error) {
+	if volumeKey.Type() == meta.Zonal && volumeKey.Zone == common.MultiZoneValue {
+		// Multi-zone support
+		// Use the node's zone.
+		volumeKey.Zone = instanceZone
+	}
+	return cloud.RepairUnderspecifiedVolumeKey(ctx, project, volumeKey)
+}
+
 // RepairUnderspecifiedVolumeKey will query the cloud provider and check each zone for the disk specified
 // by the volume key and return a volume key with a correct zone
 func (cloud *CloudProvider) RepairUnderspecifiedVolumeKey(ctx context.Context, project string, volumeKey *meta.Key) (string, *meta.Key, error) {

pkg/gce-pd-csi-driver/controller.go

@@ -47,8 +47,8 @@ type GCEControllerServer struct {
 	CloudProvider gce.GCECompute
 	Metrics       metrics.MetricsManager
 
-	disks []*compute.Disk
-	seen  map[string]int
+	volumeEntries     []*csi.ListVolumesResponse_Entry
+	volumeEntriesSeen map[string]int
 
 	snapshots      []*csi.ListSnapshotsResponse_Entry
 	snapshotTokens map[string]int
@@ -603,14 +603,24 @@ func (gceCS *GCEControllerServer) executeControllerPublishVolume(ctx context.Con
 		PublishContext: nil,
 	}
 
-	project, volKey, err = gceCS.CloudProvider.RepairUnderspecifiedVolumeKey(ctx, project, volKey)
+	instanceZone, instanceName, err := common.NodeIDToZoneAndName(nodeID)
+	if err != nil {
+		return nil, status.Errorf(codes.NotFound, "could not split nodeID: %v", err.Error()), nil
+	}
+
+	project, volKey, err = gceCS.CloudProvider.RepairUnderspecifiedVolumeKeyForZone(ctx, project, volKey, instanceZone)
 	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()), nil
 		}
 		return nil, common.LoggedError("ControllerPublishVolume error repairing underspecified volume key: ", err), nil
 	}
 
+	// Only allow read-only attachment for "multi-zone" volumes
+	if volKey.Type() == meta.Zonal && volKey.Zone == common.MultiZoneValue && !readOnly {
+		return nil, status.Errorf(codes.InvalidArgument, "'multi-zone' volume %v only supports READ_ONLY: %v", volumeID, err.Error()), nil
+	}
+
 	// 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)
@@ -625,10 +635,6 @@ func (gceCS *GCEControllerServer) executeControllerPublishVolume(ctx context.Con
 		}
 		return nil, common.LoggedError("Failed to getDisk: ", err), disk
 	}
-	instanceZone, instanceName, err := common.NodeIDToZoneAndName(nodeID)
-	if err != nil {
-		return nil, status.Errorf(codes.NotFound, "could not split nodeID: %v", err.Error()), disk
-	}
 	instance, err := gceCS.CloudProvider.GetInstanceOrError(ctx, instanceZone, instanceName)
 	if err != nil {
 		if gce.IsGCENotFoundError(err) {
@@ -739,7 +745,13 @@ func (gceCS *GCEControllerServer) executeControllerUnpublishVolume(ctx context.C
 
 	volumeID := req.GetVolumeId()
 	nodeID := req.GetNodeId()
-	project, volKey, err = gceCS.CloudProvider.RepairUnderspecifiedVolumeKey(ctx, project, volKey)
+
+	instanceZone, instanceName, err := common.NodeIDToZoneAndName(nodeID)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "could not split nodeID: %v", err.Error()), nil
+	}
+
+	project, volKey, err = gceCS.CloudProvider.RepairUnderspecifiedVolumeKeyForZone(ctx, project, volKey, instanceZone)
 	if err != nil {
 		if gce.IsGCENotFoundError(err) {
 			klog.Warningf("Treating volume %v as unpublished because it could not be found", volumeID)
@@ -756,10 +768,6 @@ func (gceCS *GCEControllerServer) executeControllerUnpublishVolume(ctx context.C
 	}
 	defer gceCS.volumeLocks.Release(lockingVolumeID)
 	diskToUnpublish, _ := gceCS.CloudProvider.GetDisk(ctx, project, volKey, gce.GCEAPIVersionV1)
-	instanceZone, instanceName, err := common.NodeIDToZoneAndName(nodeID)
-	if err != nil {
-		return nil, status.Errorf(codes.InvalidArgument, "could not split nodeID: %v", err.Error()), diskToUnpublish
-	}
 	instance, err := gceCS.CloudProvider.GetInstanceOrError(ctx, instanceZone, instanceName)
 	if err != nil {
 		if gce.IsGCENotFoundError(err) {
@@ -810,6 +818,7 @@ 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) {
@@ -879,8 +888,9 @@ func (gceCS *GCEControllerServer) ListVolumes(ctx context.Context, req *csi.List
 			"ListVolumes got max entries request %v. GCE only supports values >0", req.MaxEntries)
 	}
 
-	offset := 0
+	offsetLow := 0
 	var ok bool
+	var volumeEntries []*csi.ListVolumesResponse_Entry
 	if req.StartingToken == "" {
 		diskList, _, err := gceCS.CloudProvider.ListDisks(ctx)
 		if err != nil {
@@ -889,10 +899,14 @@ func (gceCS *GCEControllerServer) ListVolumes(ctx context.Context, req *csi.List
 			}
 			return nil, common.LoggedError("Failed to list disk: ", err)
 		}
-		gceCS.disks = diskList
-		gceCS.seen = map[string]int{}
+		volumeEntries = disksToVolumeEntries(diskList)
+	}
+
+	if req.StartingToken == "" {
+		gceCS.volumeEntries = volumeEntries
+		gceCS.volumeEntriesSeen = map[string]int{}
 	} else {
-		offset, ok = gceCS.seen[req.StartingToken]
+		offsetLow, ok = gceCS.volumeEntriesSeen[req.StartingToken]
 		if !ok {
 			return nil, status.Errorf(codes.Aborted, "ListVolumes error with invalid startingToken: %s", req.StartingToken)
 		}
@@ -903,9 +917,50 @@ func (gceCS *GCEControllerServer) ListVolumes(ctx context.Context, req *csi.List
 		maxEntries = maxListVolumesResponseEntries
 	}
 
+	nextToken := ""
+	offsetHigh := offsetLow + maxEntries
+	if offsetHigh < len(gceCS.volumeEntries) {
+		nextToken = string(uuid.NewUUID())
+		gceCS.volumeEntriesSeen[nextToken] = offsetHigh
+	} else {
+		offsetHigh = len(gceCS.volumeEntries)
+	}
+
+	return &csi.ListVolumesResponse{
+		Entries:   gceCS.volumeEntries[offsetLow:offsetHigh],
+		NextToken: nextToken,
+	}, nil
+}
+
+// isMultiZoneDisk returns the multi-zone volumeId of a disk if it is
+// "multi-zone", otherwise returns an empty string
+// The second parameter indiciates if it is a "multi-zone" disk
+func isMultiZoneDisk(diskRsrc string, diskLabels map[string]string) (string, bool) {
+	isMultiZoneDisk := false
+	for l := range diskLabels {
+		if l == common.MultiZoneLabel {
+			isMultiZoneDisk = true
+		}
+	}
+	if !isMultiZoneDisk {
+		return "", false
+	}
+
+	multiZoneVolumeId, err := common.VolumeIdAsMultiZone(diskRsrc)
+	if err != nil {
+		klog.Warningf("Error converting multi-zone volume handle for disk %s, skipped: %v", diskRsrc, err)
+		return "", false
+	}
+	return multiZoneVolumeId, true
+}
+
+// disksToVolumeEntries converts a list of disks to a list of CSI ListVolumeResponse entries
+// It appends "multi-zone" volumeHandles at the end. These are volumeHandles which
+// map to multiple volumeHandles in different zones
+func disksToVolumeEntries(disks []*compute.Disk) []*csi.ListVolumesResponse_Entry {
+	multiZoneNodesByVolumeId := map[string][]string{}
 	entries := []*csi.ListVolumesResponse_Entry{}
-	for i := 0; i+offset < len(gceCS.disks) && i < maxEntries; i++ {
-		d := gceCS.disks[i+offset]
+	for _, d := range disks {
 		diskRsrc, err := getResourceId(d.SelfLink)
 		if err != nil {
 			klog.Warningf("Bad ListVolumes disk resource %s, skipped: %v (%+v)", d.SelfLink, err, d)
@@ -920,6 +975,13 @@ func (gceCS *GCEControllerServer) ListVolumes(ctx context.Context, req *csi.List
 				users = append(users, rsrc)
 			}
 		}
+		if multiZoneVolumeId, isMultiZone := isMultiZoneDisk(diskRsrc, d.Labels); isMultiZone {
+			_, ok := multiZoneNodesByVolumeId[multiZoneVolumeId]
+			if !ok {
+				multiZoneNodesByVolumeId[multiZoneVolumeId] = []string{}
+			}
+			multiZoneNodesByVolumeId[multiZoneVolumeId] = append(multiZoneNodesByVolumeId[multiZoneVolumeId], users...)
+		}
 		entries = append(entries, &csi.ListVolumesResponse_Entry{
 			Volume: &csi.Volume{
 				VolumeId: diskRsrc,
@@ -929,17 +991,17 @@ func (gceCS *GCEControllerServer) ListVolumes(ctx context.Context, req *csi.List
 			},
 		})
 	}
-
-	nextToken := ""
-	if len(entries)+offset < len(gceCS.disks) {
-		nextToken = string(uuid.NewUUID())
-		gceCS.seen[nextToken] = len(entries) + offset
+	for volumeId, nodeIds := range multiZoneNodesByVolumeId {
+		entries = append(entries, &csi.ListVolumesResponse_Entry{
+			Volume: &csi.Volume{
+				VolumeId: volumeId,
+			},
+			Status: &csi.ListVolumesResponse_VolumeStatus{
+				PublishedNodeIds: nodeIds,
+			},
+		})
 	}
-
-	return &csi.ListVolumesResponse{
-		Entries:   entries,
-		NextToken: nextToken,
-	}, nil
+	return entries
 }
 
 func (gceCS *GCEControllerServer) GetCapacity(ctx context.Context, req *csi.GetCapacityRequest) (*csi.GetCapacityResponse, error) {

pkg/gce-pd-csi-driver/controller_test.go

@@ -67,6 +67,7 @@ var (
 
 	testVolumeID           = fmt.Sprintf("projects/%s/zones/%s/disks/%s", project, zone, name)
 	underspecifiedVolumeID = fmt.Sprintf("projects/UNSPECIFIED/zones/UNSPECIFIED/disks/%s", name)
+	multiZoneVolumeID      = fmt.Sprintf("projects/%s/zones/multi-zone/disks/%s", project, name)
 
 	region, _      = common.GetRegionFromZones([]string{zone})
 	testRegionalID = fmt.Sprintf("projects/%s/regions/%s/disks/%s", project, region, name)
@@ -3781,10 +3782,10 @@ func backoffDriver(t *testing.T, config *backoffDriverConfig) *GCEDriver {
 
 	driver := GetGCEDriver()
 	driver.cs = &GCEControllerServer{
-		Driver:       driver,
-		seen:         map[string]int{},
-		volumeLocks:  common.NewVolumeLocks(),
-		errorBackoff: newFakeCSIErrorBackoff(config.clock),
+		Driver:            driver,
+		volumeEntriesSeen: map[string]int{},
+		volumeLocks:       common.NewVolumeLocks(),
+		errorBackoff:      newFakeCSIErrorBackoff(config.clock),
 	}
 
 	driver.cs.CloudProvider = fcp

pkg/gce-pd-csi-driver/gce-pd-driver.go

@@ -156,7 +156,7 @@ func NewControllerServer(gceDriver *GCEDriver, cloudProvider gce.GCECompute, err
 	return &GCEControllerServer{
 		Driver:                 gceDriver,
 		CloudProvider:          cloudProvider,
-		seen:                   map[string]int{},
+		volumeEntriesSeen:      map[string]int{},
 		volumeLocks:            common.NewVolumeLocks(),
 		errorBackoff:           newCsiErrorBackoff(errorBackoffInitialDuration, errorBackoffMaxDuration),
 		fallbackRequisiteZones: fallbackRequisiteZones,