Add support for read_ahead_kb mount flag

Author: pwschuurman

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,48 @@ 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)
+			}
+			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()

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,29 @@ 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,
+		},
 	}
 	for _, tc := range testCases {
 		t.Logf("Test case: %s", tc.name)
 		resizeCalled := false
+		readAheadUpdateCalled := false
 		actionList := []testingexec.FakeCommandAction{
 			makeFakeCmd(
 				&testingexec.FakeCmd{
@@ -593,6 +697,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 +747,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.")
+		}
 	}
 }
 

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)
 	}
 

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

@@ -106,3 +106,22 @@ 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)
+	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
+}

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
+}