Skip to content

Commit 282f196

Browse files
tonyzhctonyzhc
committed
Derive access mode from the CreateVolume request instead of parameters.
1 parent f60f61a commit 282f196

File tree

13 files changed

+1981
-94
lines changed

13 files changed

+1981
-94
lines changed

‎pkg/common/constants.go

+5
Original file line numberDiff line numberDiff line change
@@ -32,4 +32,9 @@ const (
3232

3333
// Label that is set on a disk when it is used by a 'multi-zone' VolumeHandle
3434
MultiZoneLabel = "goog-gke-multi-zone"
35+
36+
// GCE Access Modes that are valid for hyperdisks only.
37+
GCEReadOnlyManyAccessMode = "READ_ONLY_MANY"
38+
GCEReadWriteManyAccessMode = "READ_WRITE_MANY"
39+
GCEReadWriteOnceAccessMode = "READ_WRITE_SINGLE"
3540
)

‎pkg/common/utils.go

+4
Original file line numberDiff line numberDiff line change
@@ -696,3 +696,7 @@ func NewLimiter(limit, burst int, emptyBucket bool) *rate.Limiter {
696696

697697
return limiter
698698
}
699+
700+
func IsHyperdisk(diskType string) bool {
701+
return strings.HasPrefix(diskType, "hyperdisk-")
702+
}

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

+8-39
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import (
2323

2424
"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud/meta"
2525
csi "github.com/container-storage-interface/spec/lib/go/csi"
26+
computebeta "google.golang.org/api/compute/v0.beta"
2627
computev1 "google.golang.org/api/compute/v1"
2728
"google.golang.org/api/googleapi"
2829
"google.golang.org/grpc/codes"
@@ -196,46 +197,18 @@ func (cloud *FakeCloudProvider) GetDisk(ctx context.Context, project string, vol
196197
return disk, nil
197198
}
198199

199-
func (cloud *FakeCloudProvider) ValidateExistingDisk(ctx context.Context, resp *CloudDisk, params common.DiskParameters, reqBytes, limBytes int64, multiWriter bool) error {
200-
if resp == nil {
201-
return fmt.Errorf("disk does not exist")
202-
}
203-
requestValid := common.GbToBytes(resp.GetSizeGb()) >= reqBytes || reqBytes == 0
204-
responseValid := common.GbToBytes(resp.GetSizeGb()) <= limBytes || limBytes == 0
205-
if !requestValid || !responseValid {
206-
return fmt.Errorf(
207-
"disk already exists with incompatible capacity. Need %v (Required) < %v (Existing) < %v (Limit)",
208-
reqBytes, common.GbToBytes(resp.GetSizeGb()), limBytes)
209-
}
210-
211-
respType := strings.Split(resp.GetPDType(), "/")
212-
typeMatch := strings.TrimSpace(respType[len(respType)-1]) == strings.TrimSpace(params.DiskType)
213-
typeDefault := params.DiskType == "" && strings.TrimSpace(respType[len(respType)-1]) == "pd-standard"
214-
if !typeMatch && !typeDefault {
215-
return fmt.Errorf("disk already exists with incompatible type. Need %v. Got %v",
216-
params.DiskType, respType[len(respType)-1])
217-
}
218-
219-
// We are assuming here that a multiWriter disk could be used as non-multiWriter
220-
if multiWriter && !resp.GetMultiWriter() {
221-
return fmt.Errorf("disk already exists with incompatible capability. Need MultiWriter. Got non-MultiWriter")
222-
}
223-
klog.V(4).Infof("Compatible disk already exists")
224-
return ValidateDiskParameters(resp, params)
225-
}
226-
227200
func (cloud *FakeCloudProvider) InsertDisk(ctx context.Context, project string, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool, accessMode string) error {
228201
if disk, ok := cloud.disks[volKey.String()]; ok {
229-
err := cloud.ValidateExistingDisk(ctx, disk, params,
202+
err := ValidateExistingDisk(ctx, disk, params,
230203
int64(capacityRange.GetRequiredBytes()),
231204
int64(capacityRange.GetLimitBytes()),
232-
multiWriter)
205+
multiWriter, accessMode)
233206
if err != nil {
234207
return err
235208
}
236209
}
237210

238-
computeDisk := &computev1.Disk{
211+
computeDisk := &computebeta.Disk{
239212
Name: volKey.Name,
240213
SizeGb: common.BytesToGbRoundUp(capBytes),
241214
Description: "Disk created by GCE-PD CSI Driver",
@@ -245,6 +218,8 @@ func (cloud *FakeCloudProvider) InsertDisk(ctx context.Context, project string,
245218
Labels: params.Labels,
246219
ProvisionedIops: params.ProvisionedIOPSOnCreate,
247220
ProvisionedThroughput: params.ProvisionedThroughputOnCreate,
221+
AccessMode: accessMode,
222+
MultiWriter: multiWriter,
248223
}
249224

250225
if snapshotID != "" {
@@ -263,7 +238,7 @@ func (cloud *FakeCloudProvider) InsertDisk(ctx context.Context, project string,
263238
}
264239

265240
if params.DiskEncryptionKMSKey != "" {
266-
computeDisk.DiskEncryptionKey = &computev1.CustomerEncryptionKey{
241+
computeDisk.DiskEncryptionKey = &computebeta.CustomerEncryptionKey{
267242
KmsKeyName: params.DiskEncryptionKMSKey,
268243
}
269244
}
@@ -278,13 +253,7 @@ func (cloud *FakeCloudProvider) InsertDisk(ctx context.Context, project string,
278253
return fmt.Errorf("could not create disk, key was neither zonal nor regional, instead got: %v", volKey.String())
279254
}
280255

281-
if containsBetaDiskType(hyperdiskTypes, params.DiskType) {
282-
betaDisk := convertV1DiskToBetaDisk(computeDisk)
283-
betaDisk.EnableConfidentialCompute = params.EnableConfidentialCompute
284-
cloud.disks[volKey.String()] = CloudDiskFromBeta(betaDisk)
285-
} else {
286-
cloud.disks[volKey.String()] = CloudDiskFromV1(computeDisk)
287-
}
256+
cloud.disks[volKey.String()] = CloudDiskFromBeta(computeDisk)
288257
return nil
289258
}
290259

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

+45-27
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,6 @@ const (
5353
pdDiskTypeUnsupportedPattern = `\[([a-z-]+)\] features are not compatible for creating instance`
5454
)
5555

56-
var hyperdiskTypes = []string{"hyperdisk-extreme", "hyperdisk-throughput", "hyperdisk-balanced"}
5756
var pdDiskTypeUnsupportedRegex = regexp.MustCompile(pdDiskTypeUnsupportedPattern)
5857

5958
type GCEAPIVersion string
@@ -101,7 +100,6 @@ type GCECompute interface {
101100
// Disk Methods
102101
GetDisk(ctx context.Context, project string, volumeKey *meta.Key) (*CloudDisk, error)
103102
RepairUnderspecifiedVolumeKey(ctx context.Context, project string, volumeKey *meta.Key) (string, *meta.Key, error)
104-
ValidateExistingDisk(ctx context.Context, disk *CloudDisk, params common.DiskParameters, reqBytes, limBytes int64, multiWriter bool) error
105103
InsertDisk(ctx context.Context, project string, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool, accessMode string) error
106104
DeleteDisk(ctx context.Context, project string, volumeKey *meta.Key) error
107105
UpdateDisk(ctx context.Context, project string, volKey *meta.Key, existingDisk *CloudDisk, params common.ModifyVolumeParameters) error
@@ -382,7 +380,7 @@ func (cloud *CloudProvider) getRegionURI(project, region string) string {
382380
region)
383381
}
384382

385-
func (cloud *CloudProvider) ValidateExistingDisk(ctx context.Context, resp *CloudDisk, params common.DiskParameters, reqBytes, limBytes int64, multiWriter bool) error {
383+
func ValidateExistingDisk(ctx context.Context, resp *CloudDisk, params common.DiskParameters, reqBytes, limBytes int64, multiWriter bool, accessMode string) error {
386384
klog.V(5).Infof("Validating existing disk %v with diskType: %s, reqested bytes: %v, limit bytes: %v", resp, params.DiskType, reqBytes, limBytes)
387385
if resp == nil {
388386
return fmt.Errorf("disk does not exist")
@@ -395,14 +393,33 @@ func (cloud *CloudProvider) ValidateExistingDisk(ctx context.Context, resp *Clou
395393
reqBytes, common.GbToBytes(resp.GetSizeGb()), limBytes)
396394
}
397395

398-
// We are assuming here that a multiWriter disk could be used as non-multiWriter
399-
if multiWriter && !resp.GetMultiWriter() {
396+
if common.IsHyperdisk(params.DiskType) {
397+
if validAccessMode(accessMode, resp.GetAccessMode()) {
398+
return fmt.Errorf("disk already exists with incompatible capability. Need %s. Got %s", accessMode, resp.GetAccessMode())
399+
}
400+
} else if multiWriter && !resp.GetMultiWriter() {
401+
// We are assuming here that a multiWriter PD could be used as non-multiWriter
400402
return fmt.Errorf("disk already exists with incompatible capability. Need MultiWriter. Got non-MultiWriter")
401403
}
402404

403405
return ValidateDiskParameters(resp, params)
404406
}
405407

408+
func validAccessMode(want, got string) bool {
409+
if want == got {
410+
return true
411+
}
412+
switch want {
413+
case common.GCEReadOnlyManyAccessMode:
414+
return got == common.GCEReadWriteManyAccessMode
415+
case common.GCEReadWriteOnceAccessMode:
416+
return got == common.GCEReadWriteManyAccessMode
417+
// For RWX, no other access mode is valid.
418+
default:
419+
return false
420+
}
421+
}
422+
406423
// ValidateDiskParameters takes a CloudDisk and returns true if the parameters
407424
// specified validly describe the disk provided, and false otherwise.
408425
func ValidateDiskParameters(disk *CloudDisk, params common.DiskParameters) error {
@@ -442,7 +459,7 @@ func (cloud *CloudProvider) InsertDisk(ctx context.Context, project string, volK
442459
if description == "" {
443460
description = "Regional disk created by GCE-PD CSI Driver"
444461
}
445-
return cloud.insertRegionalDisk(ctx, project, volKey, params, capBytes, capacityRange, replicaZones, snapshotID, volumeContentSourceVolumeID, description, multiWriter)
462+
return cloud.insertRegionalDisk(ctx, project, volKey, params, capBytes, capacityRange, replicaZones, snapshotID, volumeContentSourceVolumeID, description, multiWriter, accessMode)
446463
default:
447464
return fmt.Errorf("could not insert disk, key was neither zonal nor regional, instead got: %v", volKey.String())
448465
}
@@ -626,7 +643,8 @@ func (cloud *CloudProvider) insertRegionalDisk(
626643
snapshotID string,
627644
volumeContentSourceVolumeID string,
628645
description string,
629-
multiWriter bool) error {
646+
multiWriter bool,
647+
accessMode string) error {
630648
var (
631649
err error
632650
opName string
@@ -676,8 +694,13 @@ func (cloud *CloudProvider) insertRegionalDisk(
676694
}
677695
}
678696

697+
if common.IsHyperdisk(params.DiskType) {
698+
diskToCreate.AccessMode = accessMode
699+
} else {
700+
diskToCreate.MultiWriter = multiWriter
701+
}
702+
679703
var insertOp *computebeta.Operation
680-
diskToCreate.MultiWriter = multiWriter
681704
insertOp, err = cloud.betaService.RegionDisks.Insert(project, volKey.Region, diskToCreate).Context(ctx).Do()
682705
if insertOp != nil {
683706
opName = insertOp.Name
@@ -691,10 +714,10 @@ func (cloud *CloudProvider) insertRegionalDisk(
691714
// the error code should be non-Final
692715
return common.NewTemporaryError(codes.Unavailable, fmt.Errorf("error when getting disk: %w", err))
693716
}
694-
err = cloud.ValidateExistingDisk(ctx, disk, params,
717+
err = ValidateExistingDisk(ctx, disk, params,
695718
int64(capacityRange.GetRequiredBytes()),
696719
int64(capacityRange.GetLimitBytes()),
697-
multiWriter)
720+
multiWriter, accessMode)
698721
if err != nil {
699722
return err
700723
}
@@ -715,10 +738,10 @@ func (cloud *CloudProvider) insertRegionalDisk(
715738
if err != nil {
716739
return common.NewTemporaryError(codes.Unavailable, fmt.Errorf("error when getting disk: %w", err))
717740
}
718-
err = cloud.ValidateExistingDisk(ctx, disk, params,
741+
err = ValidateExistingDisk(ctx, disk, params,
719742
int64(capacityRange.GetRequiredBytes()),
720743
int64(capacityRange.GetLimitBytes()),
721-
multiWriter)
744+
multiWriter, accessMode)
722745
if err != nil {
723746
return err
724747
}
@@ -806,7 +829,12 @@ func (cloud *CloudProvider) insertZonalDisk(
806829
}
807830
}
808831

809-
diskToCreate.AccessMode = accessMode
832+
if common.IsHyperdisk(params.DiskType) {
833+
diskToCreate.AccessMode = accessMode
834+
} else {
835+
diskToCreate.MultiWriter = multiWriter
836+
}
837+
810838
var insertOp *computebeta.Operation
811839
insertOp, err = cloud.betaService.Disks.Insert(project, volKey.Zone, diskToCreate).Context(ctx).Do()
812840
if insertOp != nil {
@@ -821,10 +849,10 @@ func (cloud *CloudProvider) insertZonalDisk(
821849
// the error code should be non-Final
822850
return common.NewTemporaryError(codes.Unavailable, fmt.Errorf("error when getting disk: %w", err))
823851
}
824-
err = cloud.ValidateExistingDisk(ctx, disk, params,
852+
err = ValidateExistingDisk(ctx, disk, params,
825853
int64(capacityRange.GetRequiredBytes()),
826854
int64(capacityRange.GetLimitBytes()),
827-
multiWriter)
855+
multiWriter, accessMode)
828856
if err != nil {
829857
return err
830858
}
@@ -846,10 +874,10 @@ func (cloud *CloudProvider) insertZonalDisk(
846874
if err != nil {
847875
return common.NewTemporaryError(codes.Unavailable, fmt.Errorf("error when getting disk: %w", err))
848876
}
849-
err = cloud.ValidateExistingDisk(ctx, disk, params,
877+
err = ValidateExistingDisk(ctx, disk, params,
850878
int64(capacityRange.GetRequiredBytes()),
851879
int64(capacityRange.GetLimitBytes()),
852-
multiWriter)
880+
multiWriter, accessMode)
853881
if err != nil {
854882
return err
855883
}
@@ -1687,13 +1715,3 @@ func encodeTags(tags map[string]string) (string, error) {
16871715
}
16881716
return string(enc), nil
16891717
}
1690-
1691-
func containsBetaDiskType(betaDiskTypes []string, diskType string) bool {
1692-
for _, betaDiskType := range betaDiskTypes {
1693-
if betaDiskType == diskType {
1694-
return true
1695-
}
1696-
}
1697-
1698-
return false
1699-
}

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

+16-12
Original file line numberDiff line numberDiff line change
@@ -209,9 +209,6 @@ const (
209209
resourceProject = "projects"
210210

211211
listDisksUsersField = googleapi.Field("items/users")
212-
213-
gceReadOnlyManyAccessMode = "READ_ONLY_MANY"
214-
gceReadWriteManyAccessMode = "READ_WRITE_MANY"
215212
)
216213

217214
var (
@@ -532,12 +529,12 @@ func (gceCS *GCEControllerServer) updateAccessModeIfNecessary(ctx context.Contex
532529
return nil
533530
}
534531
project := gceCS.CloudProvider.GetDefaultProject()
535-
if disk.GetAccessMode() == gceReadOnlyManyAccessMode {
532+
if disk.GetAccessMode() == common.GCEReadOnlyManyAccessMode {
536533
// If the access mode is already readonly, return
537534
return nil
538535
}
539536

540-
return gceCS.CloudProvider.SetDiskAccessMode(ctx, project, volKey, gceReadOnlyManyAccessMode)
537+
return gceCS.CloudProvider.SetDiskAccessMode(ctx, project, volKey, common.GCEReadOnlyManyAccessMode)
541538
}
542539

543540
func (gceCS *GCEControllerServer) createSingleDeviceDisk(ctx context.Context, req *csi.CreateVolumeRequest, params common.DiskParameters) (*csi.CreateVolumeResponse, error) {
@@ -598,9 +595,17 @@ func (gceCS *GCEControllerServer) createSingleDisk(ctx context.Context, req *csi
598595
capBytes, _ := getRequestCapacity(capacityRange)
599596
multiWriter, _ := getMultiWriterFromCapabilities(req.GetVolumeCapabilities())
600597
readonly, _ := getReadOnlyFromCapabilities(req.GetVolumeCapabilities())
601-
accessMode := params.AccessMode
598+
accessMode := ""
599+
if common.IsHyperdisk(params.DiskType) {
600+
if am, err := hyperdiskAccessModeFrom(req.GetVolumeCapabilities()); err != nil {
601+
return nil, err
602+
} else {
603+
accessMode = am
604+
}
605+
}
606+
602607
if readonly && slices.Contains(disksWithModifiableAccessMode, params.DiskType) {
603-
accessMode = gceReadOnlyManyAccessMode
608+
accessMode = common.GCEReadOnlyManyAccessMode
604609
}
605610

606611
// Validate if disk already exists
@@ -613,10 +618,10 @@ func (gceCS *GCEControllerServer) createSingleDisk(ctx context.Context, req *csi
613618
}
614619
if err == nil {
615620
// There was no error so we want to validate the disk that we find
616-
err = gceCS.CloudProvider.ValidateExistingDisk(ctx, existingDisk, params,
621+
err = gce.ValidateExistingDisk(ctx, existingDisk, params,
617622
int64(capacityRange.GetRequiredBytes()),
618623
int64(capacityRange.GetLimitBytes()),
619-
multiWriter)
624+
multiWriter, accessMode)
620625
if err != nil {
621626
return nil, status.Errorf(codes.AlreadyExists, "CreateVolume disk already exists with same name and is incompatible: %v", err.Error())
622627
}
@@ -1549,9 +1554,8 @@ func (gceCS *GCEControllerServer) CreateSnapshot(ctx context.Context, req *csi.C
15491554
}
15501555
return nil, common.LoggedError("CreateSnapshot, failed to getDisk: ", err)
15511556
}
1552-
isHyperdisk := strings.HasPrefix(disk.GetPDType(), "hyperdisk-")
1553-
if isHyperdisk && disk.GetAccessMode() == gceReadWriteManyAccessMode {
1554-
return nil, status.Errorf(codes.InvalidArgument, "Cannot create snapshot for disk type %s with access mode %s", common.ParameterHdHADiskType, gceReadWriteManyAccessMode)
1557+
if common.IsHyperdisk(disk.GetPDType()) && disk.GetAccessMode() == common.GCEReadWriteManyAccessMode {
1558+
return nil, status.Errorf(codes.InvalidArgument, "Cannot create snapshot for disk type %s with access mode %s", common.ParameterHdHADiskType, common.GCEReadWriteManyAccessMode)
15551559
}
15561560

15571561
snapshotParams, err := common.ExtractAndDefaultSnapshotParameters(req.GetParameters(), gceCS.Driver.name, gceCS.Driver.extraTags)

0 commit comments

Comments
 (0)