Merge pull request #1630 from k8s-infra-cherrypick-robot/cherry-pick-1626-to-release-1.13

[release-1.13] Add flag to toggle instances.get over disks.get compute API on ControllerPublish WaitForAttach

Author: k8s-ci-robot

cmd/gce-pd-csi-driver/main.go

@@ -79,6 +79,8 @@ var (
 	computeEndpoint           *url.URL
 	allowedComputeEnvironment = []gce.Environment{gce.EnvironmentStaging, gce.EnvironmentProduction}
 
+	useInstanceAPIOnWaitForAttachDiskTypesFlag = flag.String("use-instance-api-to-poll-attachment-disk-types", "", "Comma separated list of disk types that should use instances.get API when polling for disk attach during ControllerPublish")
+
 	version string
 )
 
@@ -170,10 +172,16 @@ func handle() {
 		DiskTypes: multiZoneVolumeHandleDiskTypes,
 	}
 
+	// Initialize waitForAttach config
+	useInstanceAPIOnWaitForAttachDiskTypes := strings.Split(*useInstanceAPIOnWaitForAttachDiskTypesFlag, ",")
+	waitForAttachConfig := gce.WaitForAttachConfig{
+		UseInstancesAPIForDiskTypes: useInstanceAPIOnWaitForAttachDiskTypes,
+	}
+
 	// Initialize requirements for the controller service
 	var controllerServer *driver.GCEControllerServer
 	if *runControllerService {
-		cloudProvider, err := gce.CreateCloudProvider(ctx, version, *cloudConfigFilePath, computeEndpoint, computeEnvironment)
+		cloudProvider, err := gce.CreateCloudProvider(ctx, version, *cloudConfigFilePath, computeEndpoint, computeEnvironment, waitForAttachConfig)
 		if err != nil {
 			klog.Fatalf("Failed to get cloud provider: %v", err.Error())
 		}

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

@@ -318,7 +318,7 @@ func (cloud *FakeCloudProvider) getRegionalDiskTypeURI(project, region, diskType
 	return fmt.Sprintf(diskTypeURITemplateRegional, project, region, diskType)
 }
 
-func (cloud *FakeCloudProvider) WaitForAttach(ctx context.Context, project string, volKey *meta.Key, instanceZone, instanceName string) error {
+func (cloud *FakeCloudProvider) WaitForAttach(ctx context.Context, project string, volKey *meta.Key, diskType, instanceZone, instanceName string) error {
 	return nil
 }
 

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

@@ -99,7 +99,7 @@ type GCECompute interface {
 	DetachDisk(ctx context.Context, project, deviceName, instanceZone, instanceName string) error
 	GetDiskSourceURI(project string, volKey *meta.Key) string
 	GetDiskTypeURI(project string, volKey *meta.Key, diskType string) string
-	WaitForAttach(ctx context.Context, project string, volKey *meta.Key, instanceZone, instanceName string) error
+	WaitForAttach(ctx context.Context, project string, volKey *meta.Key, diskType, instanceZone, instanceName string) error
 	ResizeDisk(ctx context.Context, project string, volKey *meta.Key, requestBytes int64) (int64, error)
 	ListDisks(ctx context.Context) ([]*computev1.Disk, string, error)
 	// Regional Disk Methods
@@ -924,18 +924,46 @@ func (cloud *CloudProvider) waitForGlobalOp(ctx context.Context, project, opName
 	})
 }
 
-func (cloud *CloudProvider) WaitForAttach(ctx context.Context, project string, volKey *meta.Key, instanceZone, instanceName string) error {
+func (cloud *CloudProvider) waitForAttachOnInstance(ctx context.Context, project string, volKey *meta.Key, instanceZone, instanceName string) error {
 	klog.V(5).Infof("Waiting for attach of disk %v to instance %v to complete...", volKey.Name, instanceName)
 	start := time.Now()
 	return wait.ExponentialBackoff(AttachDiskBackoff, func() (bool, error) {
-		klog.V(6).Infof("Polling for attach of disk %v to instance %v to complete for %v", volKey.Name, instanceName, time.Since(start))
+		klog.V(6).Infof("Polling instances.get for attach of disk %v to instance %v to complete for %v", volKey.Name, instanceName, time.Since(start))
+		instance, err := cloud.GetInstanceOrError(ctx, instanceZone, instanceName)
+		if err != nil {
+			return false, fmt.Errorf("GetInstance failed to get instance: %w", err)
+		}
+
+		if instance == nil {
+			return false, fmt.Errorf("instance %v could not be found", instanceName)
+		}
+
+		for _, disk := range instance.Disks {
+			deviceName, err := common.GetDeviceName(volKey)
+			if err != nil {
+				return false, fmt.Errorf("failed to get disk device name for %s: %w", volKey, err)
+			}
+
+			if deviceName == disk.DeviceName {
+				return true, nil
+			}
+		}
+		return false, nil
+	})
+}
+
+func (cloud *CloudProvider) waitForAttachOnDisk(ctx context.Context, project string, volKey *meta.Key, instanceZone, instanceName string) error {
+	klog.V(5).Infof("Waiting for attach of disk %v to instance %v to complete...", volKey.Name, instanceName)
+	start := time.Now()
+	return wait.ExponentialBackoff(AttachDiskBackoff, func() (bool, error) {
+		klog.V(6).Infof("Polling disks.get for attach of disk %v to instance %v to complete for %v", volKey.Name, instanceName, time.Since(start))
 		disk, err := cloud.GetDisk(ctx, project, volKey, GCEAPIVersionV1)
 		if err != nil {
 			return false, fmt.Errorf("GetDisk failed to get disk: %w", err)
 		}
 
 		if disk == nil {
-			return false, fmt.Errorf("Disk %v could not be found", volKey.Name)
+			return false, fmt.Errorf("disk %v could not be found", volKey.Name)
 		}
 
 		for _, user := range disk.GetUsers() {
@@ -947,6 +975,14 @@ func (cloud *CloudProvider) WaitForAttach(ctx context.Context, project string, v
 	})
 }
 
+func (cloud *CloudProvider) WaitForAttach(ctx context.Context, project string, volKey *meta.Key, diskType, instanceZone, instanceName string) error {
+	if cloud.waitForAttachConfig.ShouldUseGetInstanceAPI(diskType) {
+		return cloud.waitForAttachOnInstance(ctx, project, volKey, instanceZone, instanceName)
+	} else {
+		return cloud.waitForAttachOnDisk(ctx, project, volKey, instanceZone, instanceName)
+	}
+}
+
 func wrapOpErr(name string, opErr *computev1.OperationErrorErrors) error {
 	if opErr == nil {
 		return nil

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

@@ -36,6 +36,7 @@ import (
 	"google.golang.org/api/googleapi"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/klog/v2"
+	"k8s.io/utils/strings/slices"
 )
 
 type Environment string
@@ -66,6 +67,8 @@ type CloudProvider struct {
 	zone         string
 
 	zonesCache map[string][]string
+
+	waitForAttachConfig WaitForAttachConfig
 }
 
 var _ GCECompute = &CloudProvider{}
@@ -74,14 +77,25 @@ type ConfigFile struct {
 	Global ConfigGlobal `gcfg:"global"`
 }
 
+type WaitForAttachConfig struct {
+	// A set of disk types that should use the compute instances.get API instead of the
+	// disks.get API. For certain disk types, using the instances.get API is preferred
+	// based on the response characteristics of the API.
+	UseInstancesAPIForDiskTypes []string
+}
+
+func (cfg WaitForAttachConfig) ShouldUseGetInstanceAPI(diskType string) bool {
+	return slices.Contains(cfg.UseInstancesAPIForDiskTypes, diskType)
+}
+
 type ConfigGlobal struct {
 	TokenURL  string `gcfg:"token-url"`
 	TokenBody string `gcfg:"token-body"`
 	ProjectId string `gcfg:"project-id"`
 	Zone      string `gcfg:"zone"`
 }
 
-func CreateCloudProvider(ctx context.Context, vendorVersion string, configPath string, computeEndpoint *url.URL, computeEnvironment Environment) (*CloudProvider, error) {
+func CreateCloudProvider(ctx context.Context, vendorVersion string, configPath string, computeEndpoint *url.URL, computeEnvironment Environment, waitForAttachConfig WaitForAttachConfig) (*CloudProvider, error) {
 	configFile, err := readConfig(configPath)
 	if err != nil {
 		return nil, err
@@ -120,12 +134,13 @@ func CreateCloudProvider(ctx context.Context, vendorVersion string, configPath s
 	}
 
 	return &CloudProvider{
-		service:      svc,
-		betaService:  betasvc,
-		alphaService: alphasvc,
-		project:      project,
-		zone:         zone,
-		zonesCache:   make(map[string]([]string)),
+		service:             svc,
+		betaService:         betasvc,
+		alphaService:        alphasvc,
+		project:             project,
+		zone:                zone,
+		zonesCache:          make(map[string]([]string)),
+		waitForAttachConfig: waitForAttachConfig,
 	}, nil
 
 }

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

@@ -727,7 +727,7 @@ func (gceCS *GCEControllerServer) executeControllerPublishVolume(ctx context.Con
 		return nil, common.LoggedError("Failed to Attach: ", err), disk
 	}
 
-	err = gceCS.CloudProvider.WaitForAttach(ctx, project, volKey, instanceZone, instanceName)
+	err = gceCS.CloudProvider.WaitForAttach(ctx, project, volKey, disk.GetPDType(), instanceZone, instanceName)
 	if err != nil {
 		return nil, common.LoggedError("Errored during WaitForAttach: ", err), disk
 	}

test/e2e/tests/single_zone_e2e_test.go

@@ -52,6 +52,7 @@ const (
 	defaultVolumeLimit                int64 = 127
 	readyState                              = "READY"
 	standardDiskType                        = "pd-standard"
+	ssdDiskType                             = "pd-ssd"
 	extremeDiskType                         = "pd-extreme"
 	hdtDiskType                             = "hyperdisk-throughput"
 	provisionedIOPSOnCreate                 = "12345"
@@ -297,6 +298,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 		Entry("on pd-standard", standardDiskType),
 		Entry("on pd-extreme", extremeDiskType),
 		Entry("on hyperdisk-throughput", hdtDiskType),
+		Entry("on pd-ssd", ssdDiskType),
 	)
 
 	DescribeTable("Should complete publish/unpublish lifecycle with underspecified volume ID and missing volume",
@@ -1559,6 +1561,14 @@ var typeToDisk = map[string]*disk{
 			Expect(disk.ProvisionedThroughput).To(Equal(provisionedThroughputOnCreateInt))
 		},
 	},
+	ssdDiskType: {
+		params: map[string]string{
+			common.ParameterKeyType: ssdDiskType,
+		},
+		validate: func(disk *compute.Disk) {
+			Expect(disk.Type).To(ContainSubstring(ssdDiskType))
+		},
+	},
 }
 
 func merge(a, b map[string]string) map[string]string {

test/e2e/utils/utils.go

@@ -62,7 +62,7 @@ func GCEClientAndDriverSetup(instance *remote.InstanceInfo, computeEndpoint stri
 	workspace := remote.NewWorkspaceDir("gce-pd-e2e-")
 	// Log at V(6) as the compute API calls are emitted at that level and it's
 	// useful to see what's happening when debugging tests.
-	driverRunCmd := fmt.Sprintf("sh -c '/usr/bin/nohup %s/gce-pd-csi-driver -v=6 --endpoint=%s --multi-zone-volume-handle-enable --multi-zone-volume-handle-disk-types=pd-standard %s 2> %s/prog.out < /dev/null > /dev/null &'",
+	driverRunCmd := fmt.Sprintf("sh -c '/usr/bin/nohup %s/gce-pd-csi-driver -v=6 --endpoint=%s --multi-zone-volume-handle-enable --multi-zone-volume-handle-disk-types=pd-standard --use-instance-api-to-poll-attachment-disk-types=pd-ssd %s 2> %s/prog.out < /dev/null > /dev/null &'",
 		workspace, endpoint, strings.Join(extra_flags, " "), workspace)
 
 	config := &remote.ClientConfig{