Repository: kubernetes-sigs/gcp-compute-persistent-disk-csi-driver
Pull request #1627 (2024-03-03 17:51:19)

Add support for read_ahead_kb mount flag
**What type of PR is this?**
/kind feature

**What this PR does / why we need it**:
Adds support for `read_ahead_kb` mount flag. This flag isn't passed directly to the mount command as an option, but it is used to set the readahead value for the underlying block device.

**Does this PR introduce a user-facing change?**:
```release-note
Support for read_ahead_kb mount flag to change block device readahead
```


k8s-ci-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/pull/1627#" title="Author self-approved">pwschuurman</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fgcp-compute-persistent-disk-csi-driver).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/blob/master/OWNERS)~~ [pwschuurman]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
k8s-ci-robot: @pwschuurman: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-sanity | 8f12d51e3e4f21e07ec983799a88d9b99691904c | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_gcp-compute-persistent-disk-csi-driver/1627/pull-gcp-compute-persistent-disk-csi-driver-sanity/1764347846737268736) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-sanity`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-compute-persistent-disk-csi-driver&pr=1627). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Apwschuurman). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @pwschuurman: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-gcp-compute-persistent-disk-csi-driver-unit | 16eaf30aa308433970ebec2eecafb651ca5c3bc5 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_gcp-compute-persistent-disk-csi-driver/1627/pull-gcp-compute-persistent-disk-csi-driver-unit/1764711039099736064) | true | `/test pull-gcp-compute-persistent-disk-csi-driver-unit`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-compute-persistent-disk-csi-driver&pr=1627). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Apwschuurman). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

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

Patch:
diff --git a/cmd/gce-pd-csi-driver/main.go b/cmd/gce-pd-csi-driver/main.go
index e0c8daf2..2b8e4bbb 100644
--- a/cmd/gce-pd-csi-driver/main.go
+++ b/cmd/gce-pd-csi-driver/main.go
@@ -79,8 +79,6 @@ 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
 )
 
@@ -172,16 +170,10 @@ 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, waitForAttachConfig)
+		cloudProvider, err := gce.CreateCloudProvider(ctx, version, *cloudConfigFilePath, computeEndpoint, computeEnvironment)
 		if err != nil {
 			klog.Fatalf("Failed to get cloud provider: %v", err.Error())
 		}
diff --git a/pkg/gce-cloud-provider/compute/fake-gce.go b/pkg/gce-cloud-provider/compute/fake-gce.go
index b2ed358e..ced0d26e 100644
--- a/pkg/gce-cloud-provider/compute/fake-gce.go
+++ b/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, diskType, instanceZone, instanceName string) error {
+func (cloud *FakeCloudProvider) WaitForAttach(ctx context.Context, project string, volKey *meta.Key, instanceZone, instanceName string) error {
 	return nil
 }
 
diff --git a/pkg/gce-cloud-provider/compute/gce-compute.go b/pkg/gce-cloud-provider/compute/gce-compute.go
index a35a948d..d71f4b4b 100644
--- a/pkg/gce-cloud-provider/compute/gce-compute.go
+++ b/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, diskType, instanceZone, instanceName string) error
+	WaitForAttach(ctx context.Context, project string, volKey *meta.Key, 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,46 +924,18 @@ func (cloud *CloudProvider) waitForGlobalOp(ctx context.Context, project, opName
 	})
 }
 
-func (cloud *CloudProvider) waitForAttachOnInstance(ctx context.Context, project string, volKey *meta.Key, instanceZone, instanceName string) error {
+func (cloud *CloudProvider) WaitForAttach(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 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))
+		klog.V(6).Infof("Polling 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() {
@@ -975,14 +947,6 @@ func (cloud *CloudProvider) waitForAttachOnDisk(ctx context.Context, project str
 	})
 }
 
-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
diff --git a/pkg/gce-cloud-provider/compute/gce.go b/pkg/gce-cloud-provider/compute/gce.go
index 10494385..2c769ced 100644
--- a/pkg/gce-cloud-provider/compute/gce.go
+++ b/pkg/gce-cloud-provider/compute/gce.go
@@ -36,7 +36,6 @@ import (
 	"google.golang.org/api/googleapi"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/klog/v2"
-	"k8s.io/utils/strings/slices"
 )
 
 type Environment string
@@ -67,8 +66,6 @@ type CloudProvider struct {
 	zone         string
 
 	zonesCache map[string][]string
-
-	waitForAttachConfig WaitForAttachConfig
 }
 
 var _ GCECompute = &CloudProvider{}
@@ -77,17 +74,6 @@ 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"`
@@ -95,7 +81,7 @@ type ConfigGlobal struct {
 	Zone      string `gcfg:"zone"`
 }
 
-func CreateCloudProvider(ctx context.Context, vendorVersion string, configPath string, computeEndpoint *url.URL, computeEnvironment Environment, waitForAttachConfig WaitForAttachConfig) (*CloudProvider, error) {
+func CreateCloudProvider(ctx context.Context, vendorVersion string, configPath string, computeEndpoint *url.URL, computeEnvironment Environment) (*CloudProvider, error) {
 	configFile, err := readConfig(configPath)
 	if err != nil {
 		return nil, err
@@ -134,13 +120,12 @@ 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)),
-		waitForAttachConfig: waitForAttachConfig,
+		service:      svc,
+		betaService:  betasvc,
+		alphaService: alphasvc,
+		project:      project,
+		zone:         zone,
+		zonesCache:   make(map[string]([]string)),
 	}, nil
 
 }
diff --git a/pkg/gce-pd-csi-driver/controller.go b/pkg/gce-pd-csi-driver/controller.go
index 73369405..486b400d 100644
--- a/pkg/gce-pd-csi-driver/controller.go
+++ b/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, disk.GetPDType(), instanceZone, instanceName)
+	err = gceCS.CloudProvider.WaitForAttach(ctx, project, volKey, instanceZone, instanceName)
 	if err != nil {
 		return nil, common.LoggedError("Errored during WaitForAttach: ", err), disk
 	}
diff --git a/pkg/gce-pd-csi-driver/node.go b/pkg/gce-pd-csi-driver/node.go
index e099ef7d..0034ee9c 100644
--- a/pkg/gce-pd-csi-driver/node.go
+++ b/pkg/gce-pd-csi-driver/node.go
@@ -19,7 +19,9 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"regexp"
 	"runtime"
+	"strconv"
 	"time"
 
 	"google.golang.org/grpc/codes"
@@ -72,6 +74,12 @@ const (
 	defaultLinuxFsType         = "ext4"
 	defaultWindowsFsType       = "ntfs"
 	fsTypeExt3                 = "ext3"
+
+	readAheadKBMountFlagRegexPattern = "^read_ahead_kb=(.+)$"
+)
+
+var (
+	readAheadKBMountFlagRegex = regexp.MustCompile(readAheadKBMountFlagRegexPattern)
 )
 
 func getDefaultFsType() string {
@@ -318,12 +326,19 @@ func (ns *GCENodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStage
 	// Part 3: Mount device to stagingTargetPath
 	fstype := getDefaultFsType()
 
+	shouldUpdateReadAhead := false
+	var readAheadKB int64
 	options := []string{}
 	if mnt := volumeCapability.GetMount(); mnt != nil {
 		if mnt.FsType != "" {
 			fstype = mnt.FsType
 		}
 		options = collectMountOptions(fstype, mnt.MountFlags)
+
+		readAheadKB, shouldUpdateReadAhead, err = extractReadAheadKBMountFlag(mnt.MountFlags)
+		if err != nil {
+			return nil, status.Errorf(codes.InvalidArgument, "failure parsing mount flags: %v", err.Error())
+		}
 	} else if blk := volumeCapability.GetBlock(); blk != nil {
 		// Noop for Block NodeStageVolume
 		klog.V(4).Infof("NodeStageVolume succeeded on %v to %s, capability is block so this is a no-op", volumeID, stagingTargetPath)
@@ -368,10 +383,53 @@ func (ns *GCENodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStage
 		}
 	}
 
+	// Part 5: Update read_ahead
+	if shouldUpdateReadAhead {
+		if err := ns.updateReadAhead(devicePath, readAheadKB); err != nil {
+			return nil, status.Errorf(codes.Internal, "failure updating readahead for %s to %dKB: %v", devicePath, readAheadKB, err.Error())
+		}
+	}
+
 	klog.V(4).Infof("NodeStageVolume succeeded on %v to %s", volumeID, stagingTargetPath)
 	return &csi.NodeStageVolumeResponse{}, nil
 }
 
+func (ns *GCENodeServer) updateReadAhead(devicePath string, readAheadKB int64) error {
+	isBlock, err := ns.VolumeStatter.IsBlockDevice(devicePath)
+	if err != nil {
+		return fmt.Errorf("failed to determine whether %s is a block device: %v", devicePath, err)
+	}
+	if !isBlock {
+		return nil
+	}
+
+	if err := setReadAheadKB(devicePath, readAheadKB, ns.Mounter); err != nil {
+		return fmt.Errorf("failed to set readahead: %v", err)
+	}
+
+	return nil
+}
+
+func extractReadAheadKBMountFlag(mountFlags []string) (int64, bool, error) {
+	for _, mountFlag := range mountFlags {
+		if readAheadKB := readAheadKBMountFlagRegex.FindStringSubmatch(mountFlag); len(readAheadKB) == 2 {
+			// There is only one matching pattern in readAheadKBMountFlagRegex
+			// If found, it will be at index 1
+			readAheadKBInt, err := strconv.ParseInt(readAheadKB[1], 10, 0)
+			if err != nil {
+				return -1, false, fmt.Errorf("invalid read_ahead_kb mount flag %q: %v", mountFlag, err)
+			}
+			if readAheadKBInt < 0 {
+				// Negative values can result in unintuitive values when setting read ahead
+				// (due to blockdev intepreting negative integers as large positive integers).
+				return -1, false, fmt.Errorf("invalid negative value for read_ahead_kb mount flag: %q", mountFlag)
+			}
+			return readAheadKBInt, true, nil
+		}
+	}
+	return -1, false, nil
+}
+
 func (ns *GCENodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstageVolumeRequest) (*csi.NodeUnstageVolumeResponse, error) {
 	// Validate arguments
 	volumeID := req.GetVolumeId()
diff --git a/pkg/gce-pd-csi-driver/utils.go b/pkg/gce-pd-csi-driver/utils.go
index a33d66cc..8e14b272 100644
--- a/pkg/gce-pd-csi-driver/utils.go
+++ b/pkg/gce-pd-csi-driver/utils.go
@@ -257,6 +257,11 @@ func collectMountOptions(fsType string, mntFlags []string) []string {
 	var options []string
 
 	for _, opt := range mntFlags {
+		if readAheadKBMountFlagRegex.FindString(opt) != "" {
+			// The read_ahead_kb flag is a special flag that isn't
+			// passed directly as an option to the mount command.
+			continue
+		}
 		options = append(options, opt)
 	}
 
diff --git a/pkg/gce-pd-csi-driver/utils_linux.go b/pkg/gce-pd-csi-driver/utils_linux.go
index 805fb6e8..a8ce16da 100644
--- a/pkg/gce-pd-csi-driver/utils_linux.go
+++ b/pkg/gce-pd-csi-driver/utils_linux.go
@@ -106,3 +106,24 @@ func getBlockSizeBytes(devicePath string, m *mount.SafeFormatAndMount) (int64, e
 	}
 	return gotSizeBytes, nil
 }
+
+func setReadAheadKB(devicePath string, readAheadKB int64, m *mount.SafeFormatAndMount) error {
+	output, err := m.Exec.Command("blockdev", "--getss", devicePath).CombinedOutput()
+	if err != nil {
+		return fmt.Errorf("error when reading sector size at path %s: output: %s, err: %w", devicePath, string(output), err)
+	}
+	strOut := strings.TrimSpace(string(output))
+	sectorSizeBytes, err := strconv.ParseInt(strOut, 10, 64)
+	if err != nil {
+		return fmt.Errorf("failed to parse %q into an int size", strOut)
+	}
+	readAheadInSectors := readAheadKB * 1024 / sectorSizeBytes
+	readAheadInSectorsStr := strconv.FormatInt(readAheadInSectors, 10)
+	// Empirical testing indicates that the actual read_ahead_kb size that is set is rounded to the
+	// nearest 4KB.
+	output, err = m.Exec.Command("blockdev", "--setra", readAheadInSectorsStr, devicePath).CombinedOutput()
+	if err != nil {
+		return fmt.Errorf("error when setting readahead at path %s: output: %s, err: %w", devicePath, string(output), err)
+	}
+	return nil
+}
diff --git a/pkg/gce-pd-csi-driver/utils_windows.go b/pkg/gce-pd-csi-driver/utils_windows.go
index b12f4f82..f40639d4 100644
--- a/pkg/gce-pd-csi-driver/utils_windows.go
+++ b/pkg/gce-pd-csi-driver/utils_windows.go
@@ -105,3 +105,8 @@ func getBlockSizeBytes(devicePath string, m *mount.SafeFormatAndMount) (int64, e
 	}
 	return proxy.GetDiskTotalBytes(devicePath)
 }
+
+func setReadAheadKB(devicePath string, readAheadKB int64, m *mount.SafeFormatAndMount) error {
+	// This is a no-op on windows.
+	return nil
+}
diff --git a/pkg/mount-manager/statter_linux.go b/pkg/mount-manager/statter_linux.go
index d75c39b3..921ffdb9 100644
--- a/pkg/mount-manager/statter_linux.go
+++ b/pkg/mount-manager/statter_linux.go
@@ -61,10 +61,26 @@ func (*realStatter) StatFS(path string) (available, capacity, used, inodesFree,
 	return
 }
 
-type fakeStatter struct{}
+type fakeStatter struct {
+	options FakeStatterOptions
+}
+
+type FakeStatterOptions struct {
+	IsBlock bool
+}
 
 func NewFakeStatter(mounter *mount.SafeFormatAndMount) *fakeStatter {
-	return &fakeStatter{}
+	return &fakeStatter{
+		options: FakeStatterOptions{
+			IsBlock: true,
+		},
+	}
+}
+
+func NewFakeStatterWithOptions(mounter *mount.SafeFormatAndMount, options FakeStatterOptions) *fakeStatter {
+	return &fakeStatter{
+		options: options,
+	}
 }
 
 func (*fakeStatter) StatFS(path string) (available, capacity, used, inodesFree, inodes, inodesUsed int64, err error) {
@@ -72,6 +88,6 @@ func (*fakeStatter) StatFS(path string) (available, capacity, used, inodesFree,
 	return 1, 1, 1, 1, 1, 1, nil
 }
 
-func (*fakeStatter) IsBlockDevice(fullPath string) (bool, error) {
-	return false, nil
+func (fs *fakeStatter) IsBlockDevice(fullPath string) (bool, error) {
+	return fs.options.IsBlock, nil
 }


Test patch:
diff --git a/pkg/gce-pd-csi-driver/node_test.go b/pkg/gce-pd-csi-driver/node_test.go
index ce774533..a50c7464 100644
--- a/pkg/gce-pd-csi-driver/node_test.go
+++ b/pkg/gce-pd-csi-driver/node_test.go
@@ -28,6 +28,7 @@ import (
 	testingexec "k8s.io/utils/exec/testing"
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/google/go-cmp/cmp"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"k8s.io/mount-utils"
@@ -109,21 +110,33 @@ func TestNodeGetVolumeStats(t *testing.T) {
 		Readonly:          false,
 		VolumeCapability:  stdVolCap,
 	}
+
 	_, err = ns.NodePublishVolume(context.Background(), req)
 	if err != nil {
 		t.Fatalf("Failed to set up test by publishing default vol: %v", err)
 	}
 
 	testCases := []struct {
-		name       string
-		volumeID   string
-		volumePath string
-		expectErr  bool
+		name           string
+		volumeID       string
+		volumePath     string
+		expectedResp   *csi.NodeGetVolumeStatsResponse
+		deviceCapacity int
+		expectErr      bool
 	}{
 		{
-			name:       "normal",
-			volumeID:   defaultVolumeID,
-			volumePath: targetPath,
+			name:           "normal",
+			volumeID:       defaultVolumeID,
+			volumePath:     targetPath,
+			deviceCapacity: 300 * 1024 * 1024 * 1024, // 300 GB
+			expectedResp: &csi.NodeGetVolumeStatsResponse{
+				Usage: []*csi.VolumeUsage{
+					{
+						Unit:  csi.VolumeUsage_BYTES,
+						Total: 300 * 1024 * 1024 * 1024, // 300 GB,
+					},
+				},
+			},
 		},
 		{
 			name:       "no vol id",
@@ -145,18 +158,38 @@ func TestNodeGetVolumeStats(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
+			actionList := []testingexec.FakeCommandAction{
+				makeFakeCmd(
+					&testingexec.FakeCmd{
+						CombinedOutputScript: []testingexec.FakeAction{
+							func() ([]byte, []byte, error) {
+								return []byte(fmt.Sprintf("%d", tc.deviceCapacity)), nil, nil
+							},
+						},
+					},
+					"blockdev",
+					strings.Split("--getsize64 /dev/disk/fake-path", " ")...,
+				),
+			}
+
+			mounter := mountmanager.NewFakeSafeMounterWithCustomExec(&testingexec.FakeExec{CommandScript: actionList})
+			gceDriver := getTestGCEDriverWithCustomMounter(t, mounter)
+			ns := gceDriver.ns
 
 			req := &csi.NodeGetVolumeStatsRequest{
 				VolumeId:   tc.volumeID,
 				VolumePath: tc.volumePath,
 			}
-			_, err := ns.NodeGetVolumeStats(context.Background(), req)
+			resp, err := ns.NodeGetVolumeStats(context.Background(), req)
 			if err != nil && !tc.expectErr {
 				t.Fatalf("Got unexpected err: %v", err)
 			}
 			if err == nil && tc.expectErr {
 				t.Fatal("Did not get error but expected one")
 			}
+			if diff := cmp.Diff(tc.expectedResp, resp); diff != "" {
+				t.Errorf("NodeGetVolumeStats(%s): -want, +got \n%s", req, diff)
+			}
 		})
 	}
 }
@@ -394,13 +427,17 @@ func TestNodeStageVolume(t *testing.T) {
 	stagingPath := filepath.Join(tempDir, defaultStagingPath)
 
 	testCases := []struct {
-		name         string
-		req          *csi.NodeStageVolumeRequest
-		deviceSize   int
-		blockExtSize int
-		readonlyBit  string
-		expResize    bool
-		expErrCode   codes.Code
+		name               string
+		req                *csi.NodeStageVolumeRequest
+		deviceSize         int
+		blockExtSize       int
+		readonlyBit        string
+		expResize          bool
+		expReadAheadUpdate bool
+		expReadAheadKB     string
+		readAheadSectors   string
+		sectorSizeInBytes  int
+		expErrCode         codes.Code
 	}{
 		{
 			name: "Valid request, no resize because block and filesystem sizes match",
@@ -450,6 +487,54 @@ func TestNodeStageVolume(t *testing.T) {
 			readonlyBit:  "0",
 			expResize:    false,
 		},
+		{
+			name: "Valid request, update readahead",
+			req: &csi.NodeStageVolumeRequest{
+				VolumeId:          volumeID,
+				StagingTargetPath: stagingPath,
+				VolumeCapability: &csi.VolumeCapability{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{
+							MountFlags: []string{"read_ahead_kb=4096"},
+						},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+					},
+				},
+			},
+			deviceSize:         5,
+			blockExtSize:       1,
+			readonlyBit:        "0",
+			expResize:          true,
+			expReadAheadUpdate: true,
+			readAheadSectors:   "8192",
+			sectorSizeInBytes:  512,
+		},
+		{
+			name: "Valid request, update readahead (different sectorsize)",
+			req: &csi.NodeStageVolumeRequest{
+				VolumeId:          volumeID,
+				StagingTargetPath: stagingPath,
+				VolumeCapability: &csi.VolumeCapability{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{
+							MountFlags: []string{"read_ahead_kb=4096"},
+						},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+					},
+				},
+			},
+			deviceSize:         5,
+			blockExtSize:       1,
+			readonlyBit:        "0",
+			expResize:          true,
+			expReadAheadUpdate: true,
+			readAheadSectors:   "4194304",
+			sectorSizeInBytes:  1,
+		},
 		{
 			name: "Invalid request (Bad Access Mode)",
 			req: &csi.NodeStageVolumeRequest{
@@ -493,10 +578,47 @@ func TestNodeStageVolume(t *testing.T) {
 			},
 			expErrCode: codes.InvalidArgument,
 		},
+		{
+			name: "Invalid request (Invalid read_ahead_kb)",
+			req: &csi.NodeStageVolumeRequest{
+				VolumeId:          volumeID,
+				StagingTargetPath: stagingPath,
+				VolumeCapability: &csi.VolumeCapability{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{
+							MountFlags: []string{"read_ahead_kb=not_a_number"},
+						},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+					},
+				},
+			},
+			expErrCode: codes.InvalidArgument,
+		},
+		{
+			name: "Invalid request (negative read_ahead_kb)",
+			req: &csi.NodeStageVolumeRequest{
+				VolumeId:          volumeID,
+				StagingTargetPath: stagingPath,
+				VolumeCapability: &csi.VolumeCapability{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{
+							MountFlags: []string{"read_ahead_kb=-4096"},
+						},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+					},
+				},
+			},
+			expErrCode: codes.InvalidArgument,
+		},
 	}
 	for _, tc := range testCases {
 		t.Logf("Test case: %s", tc.name)
 		resizeCalled := false
+		readAheadUpdateCalled := false
 		actionList := []testingexec.FakeCommandAction{
 			makeFakeCmd(
 				&testingexec.FakeCmd{
@@ -593,6 +715,33 @@ func TestNodeStageVolume(t *testing.T) {
 				),
 			}...)
 		}
+		if tc.expReadAheadUpdate {
+			actionList = append(actionList, []testingexec.FakeCommandAction{
+				makeFakeCmd(
+					&testingexec.FakeCmd{
+						CombinedOutputScript: []testingexec.FakeAction{
+							func() ([]byte, []byte, error) {
+								return []byte(fmt.Sprintf("%d", tc.sectorSizeInBytes)), nil, nil
+							},
+						},
+					},
+					"blockdev",
+					[]string{"--getss", "/dev/disk/fake-path"}...,
+				),
+				makeFakeCmd(
+					&testingexec.FakeCmd{
+						CombinedOutputScript: []testingexec.FakeAction{
+							func() (_ []byte, args []byte, _ error) {
+								readAheadUpdateCalled = true
+								return []byte{}, nil, nil
+							},
+						},
+					},
+					"blockdev",
+					[]string{"--setra", tc.readAheadSectors, "/dev/disk/fake-path"}...,
+				),
+			}...)
+		}
 		mounter := mountmanager.NewFakeSafeMounterWithCustomExec(&testingexec.FakeExec{CommandScript: actionList})
 		gceDriver := getTestGCEDriverWithCustomMounter(t, mounter)
 		ns := gceDriver.ns
@@ -616,6 +765,12 @@ func TestNodeStageVolume(t *testing.T) {
 		if tc.expResize == false && resizeCalled == true {
 			t.Fatalf("Test called resize, but it was not expected.")
 		}
+		if tc.expReadAheadUpdate == true && readAheadUpdateCalled == false {
+			t.Fatalf("Test did not update read ahead, but it was expected.")
+		}
+		if tc.expReadAheadUpdate == false && readAheadUpdateCalled == true {
+			t.Fatalf("Test updated read ahead, but it was not expected.")
+		}
 	}
 }
 
diff --git a/test/e2e/tests/single_zone_e2e_test.go b/test/e2e/tests/single_zone_e2e_test.go
index f9c096fe..9c4c7d38 100644
--- a/test/e2e/tests/single_zone_e2e_test.go
+++ b/test/e2e/tests/single_zone_e2e_test.go
@@ -52,7 +52,6 @@ const (
 	defaultVolumeLimit                int64 = 127
 	readyState                              = "READY"
 	standardDiskType                        = "pd-standard"
-	ssdDiskType                             = "pd-ssd"
 	extremeDiskType                         = "pd-extreme"
 	hdtDiskType                             = "hyperdisk-throughput"
 	provisionedIOPSOnCreate                 = "12345"
@@ -298,7 +297,6 @@ 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",
@@ -1317,6 +1315,90 @@ var _ = Describe("GCE PD CSI Driver", func() {
 		Expect(err).To(BeNil(), "no error expected when passed valid compute url")
 	})
 
+	It("Should update readahead if read_ahead_kb passed on mount", func() {
+		testContext := getRandomTestContext()
+
+		p, z, _ := testContext.Instance.GetIdentity()
+		client := testContext.Client
+		instance := testContext.Instance
+
+		// Create Disk
+		volName, volID := createAndValidateUniqueZonalDisk(client, p, z, standardDiskType)
+
+		defer func() {
+			// Delete Disk
+			err := client.DeleteVolume(volID)
+			Expect(err).To(BeNil(), "DeleteVolume failed")
+
+			// Validate Disk Deleted
+			_, err = computeService.Disks.Get(p, z, volName).Do()
+			Expect(gce.IsGCEError(err, "notFound")).To(BeTrue(), "Expected disk to not be found")
+		}()
+
+		// Attach Disk
+		err := client.ControllerPublishVolumeReadWrite(volID, instance.GetNodeID(), false /* forceAttach */)
+		Expect(err).To(BeNil(), "ControllerPublishVolume failed with error for disk %v on node %v: %v", volID, instance.GetNodeID(), err)
+
+		defer func() {
+			// Detach Disk
+			err = client.ControllerUnpublishVolume(volID, instance.GetNodeID())
+			if err != nil {
+				klog.Errorf("Failed to detach disk: %v", err)
+			}
+
+		}()
+
+		// Stage Disk
+		stageDir := filepath.Join("/tmp/", volName, "stage")
+		expectedReadAheadKB := "4096"
+		volCap := &csi.VolumeCapability{
+			AccessType: &csi.VolumeCapability_Mount{
+				Mount: &csi.VolumeCapability_MountVolume{
+					MountFlags: []string{fmt.Sprintf("read_ahead_kb=%s", expectedReadAheadKB)},
+				},
+			},
+			AccessMode: &csi.VolumeCapability_AccessMode{
+				Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+			},
+		}
+		err = client.NodeStageVolume(volID, stageDir, volCap)
+		Expect(err).To(BeNil(), "failed to stage volume: %v", err)
+
+		// Validate that the link is correct
+		var validated bool
+		var devName string
+		devicePaths := deviceutils.NewDeviceUtils().GetDiskByIdPaths(volName, "")
+		for _, devicePath := range devicePaths {
+			validated, err = testutils.ValidateLogicalLinkIsDisk(instance, devicePath, volName)
+			Expect(err).To(BeNil(), "failed to validate link %s is disk %s: %v", stageDir, volName, err)
+			if validated {
+				devFsPath, err := instance.SSH("find", devicePath, "-printf", "'%l'")
+				Expect(err).To(BeNil(), "Failed to symlink devicePath")
+				devFsPathPieces := strings.Split(devFsPath, "/")
+				devName = devFsPathPieces[len(devFsPathPieces)-1]
+
+			}
+		}
+		Expect(validated).To(BeTrue(), "could not find device in %v that links to volume %s", devicePaths, volName)
+		actualReadAheadKBStr, err := instance.SSH("cat", fmt.Sprintf("/sys/block/%s/queue/read_ahead_kb", devName))
+		actualReadAheadKB := strings.TrimSpace(actualReadAheadKBStr)
+		Expect(err).To(BeNil(), "Failed to read read_ahead_kb: %v", err)
+		Expect(actualReadAheadKB).To(Equal(expectedReadAheadKB), "unexpected read_ahead_kb")
+
+		defer func() {
+			// Unstage Disk
+			err = client.NodeUnstageVolume(volID, stageDir)
+			if err != nil {
+				klog.Errorf("Failed to unstage volume: %v", err)
+			}
+			fp := filepath.Join("/tmp/", volName)
+			err = testutils.RmAll(instance, fp)
+			if err != nil {
+				klog.Errorf("Failed to rm file path %s: %v", fp, err)
+			}
+		}()
+	})
+
 	type multiZoneTestConfig struct {
 		diskType          string
 		readOnly          bool
@@ -1561,14 +1643,6 @@ 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 {
diff --git a/test/e2e/utils/utils.go b/test/e2e/utils/utils.go
index 00e8bb65..79f59470 100644
--- a/test/e2e/utils/utils.go
+++ b/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 --use-instance-api-to-poll-attachment-disk-types=pd-ssd %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 %s 2> %s/prog.out < /dev/null > /dev/null &'",
 		workspace, endpoint, strings.Join(extra_flags, " "), workspace)
 	config := &remote.ClientConfig{
 		PkgPath:      pkgPath,
diff --git a/test/sanity/sanity_test.go b/test/sanity/sanity_test.go
index c093125d..38036535 100644
--- a/test/sanity/sanity_test.go
+++ b/test/sanity/sanity_test.go
@@ -71,7 +71,8 @@ func TestSanity(t *testing.T) {
 	//Initialize GCE Driver
 	identityServer := driver.NewIdentityServer(gceDriver)
 	controllerServer := driver.NewControllerServer(gceDriver, cloudProvider, 0, 5*time.Minute, fallbackRequisiteZones, enableStoragePools, multiZoneVolumeHandleConfig)
-	nodeServer := driver.NewNodeServer(gceDriver, mounter, deviceUtils, metadataservice.NewFakeService(), mountmanager.NewFakeStatter(mounter))
+	fakeStatter := mountmanager.NewFakeStatterWithOptions(mounter, mountmanager.FakeStatterOptions{IsBlock: false})
+	nodeServer := driver.NewNodeServer(gceDriver, mounter, deviceUtils, metadataservice.NewFakeService(), fakeStatter)
 	err = gceDriver.SetupGCEDriver(driverName, vendorVersion, extraLabels, identityServer, controllerServer, nodeServer)
 	if err != nil {
 		t.Fatalf("Failed to initialize GCE CSI Driver: %v", err.Error())