Merge branch 'master' into logs-for-device-mappings

Author: cemakd

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

@@ -100,6 +100,8 @@ var (
 
 	diskCacheSyncPeriod = flag.Duration("disk-cache-sync-period", 10*time.Minute, "Period for the disk cache to check the /dev/disk/by-id/ directory and evaluate the symlinks")
 
+	enableDiskSizeValidation = flag.Bool("enable-disk-size-validation", false, "If set to true, the driver will validate that the requested disk size is matches the physical disk size. This flag is disabled by default.")
+
 	version string
 )
 
@@ -251,7 +253,8 @@ func handle() {
 		maxBackoffDuration := time.Duration(*errorBackoffMaxDurationMs) * time.Millisecond
 		// TODO(2042): Move more of the constructor args into this struct
 		args := &driver.GCEControllerServerArgs{
-			EnableDiskTopology: *diskTopology,
+			EnableDiskTopology:       *diskTopology,
+			EnableDiskSizeValidation: *enableDiskSizeValidation,
 		}
 
 		controllerServer = driver.NewControllerServer(gceDriver, cloudProvider, initialBackoffDuration, maxBackoffDuration, fallbackRequisiteZones, *enableStoragePoolsFlag, *enableDataCacheFlag, multiZoneVolumeHandleConfig, listVolumesConfig, provisionableDisksConfig, *enableHdHAFlag, args)

pkg/common/constants.go

@@ -48,6 +48,7 @@ const (
 
 	ContextDataCacheSize = "data-cache-size"
 	ContextDataCacheMode = "data-cache-mode"
+	ContextDiskSizeGB    = "disk-size"
 
 	// Keys in the publish context
 	ContexLocalSsdCacheSize = "local-ssd-cache-size"

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

@@ -21,6 +21,7 @@ import (
 	"math/rand"
 	neturl "net/url"
 	"sort"
+	"strconv"
 	"strings"
 	"time"
 
@@ -121,11 +122,13 @@ type GCEControllerServer struct {
 	// new RPC methods that might be introduced in future versions of the spec.
 	csi.UnimplementedControllerServer
 
-	EnableDiskTopology bool
+	EnableDiskTopology       bool
+	EnableDiskSizeValidation bool
 }
 
 type GCEControllerServerArgs struct {
-	EnableDiskTopology bool
+	EnableDiskTopology       bool
+	EnableDiskSizeValidation bool
 }
 
 type MultiZoneVolumeHandleConfig struct {
@@ -1113,7 +1116,7 @@ func (gceCS *GCEControllerServer) executeControllerPublishVolume(ctx context.Con
 	volumeCapability := req.GetVolumeCapability()
 
 	pubVolResp := &csi.ControllerPublishVolumeResponse{
-		PublishContext: nil,
+		PublishContext: map[string]string{},
 	}
 
 	// Set data cache publish context
@@ -1162,6 +1165,9 @@ func (gceCS *GCEControllerServer) executeControllerPublishVolume(ctx context.Con
 		}
 		return nil, common.LoggedError("Failed to getDisk: ", err), disk
 	}
+	if gceCS.EnableDiskSizeValidation && pubVolResp.GetPublishContext() != nil {
+		pubVolResp.PublishContext[common.ContextDiskSizeGB] = strconv.FormatInt(disk.GetSizeGb(), 10)
+	}
 	instance, err := gceCS.CloudProvider.GetInstanceOrError(ctx, project, instanceZone, instanceName)
 	if err != nil {
 		if gce.IsGCENotFoundError(err) {

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

@@ -178,6 +178,7 @@ func NewControllerServer(gceDriver *GCEDriver, cloudProvider gce.GCECompute, err
 		provisionableDisksConfig:    provisionableDisksConfig,
 		enableHdHA:                  enableHdHA,
 		EnableDiskTopology:          args.EnableDiskTopology,
+		EnableDiskSizeValidation:    args.EnableDiskSizeValidation,
 	}
 }
 

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

@@ -449,6 +449,17 @@ func (ns *GCENodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStage
 		klog.V(4).Infof("CSI volume is read-only, mounting with extra option ro")
 	}
 
+	// If a disk size is provided in the publish context, ensure it matches the actual device size.
+	if expectedSize := req.GetPublishContext()[common.ContextDiskSizeGB]; expectedSize != "" {
+		actualSize, err := getBlockSizeBytes(devicePath, ns.Mounter)
+		if err != nil {
+			return nil, status.Error(codes.Internal, fmt.Sprintf("failed to get block size for '%s': %v", devicePath, err.Error()))
+		}
+		if expectedSize != strconv.FormatInt(actualSize, 10) {
+			return nil, status.Error(codes.Internal, fmt.Sprintf("expected block size %q, got %q", expectedSize, strconv.FormatInt(actualSize, 10)))
+		}
+	}
+
 	err = ns.formatAndMount(devicePath, stagingTargetPath, fstype, options, ns.Mounter)
 	if err != nil {
 		// If a volume is created from a content source like snapshot or cloning, the filesystem might get marked

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

@@ -33,6 +33,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"k8s.io/mount-utils"
+	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/deviceutils"
 	metadataservice "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/metadata"
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/linkcache"
@@ -1123,6 +1124,41 @@ func TestNodeStageVolume(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "Valid request with disk size check",
+			req: &csi.NodeStageVolumeRequest{
+				VolumeId:          volumeID,
+				StagingTargetPath: stagingPath,
+				VolumeCapability:  stdVolCap,
+				PublishContext:    map[string]string{common.ContextDiskSizeGB: "1"},
+			},
+			deviceSize:   1,
+			blockExtSize: 1,
+			readonlyBit:  "1",
+			expResize:    false,
+			expCommandList: []fakeCmd{
+				{
+					cmd:    "blockdev",
+					args:   "--getsize64 /dev/disk/fake-path",
+					stdout: "%v",
+				},
+				{
+					cmd:    "blkid",
+					args:   "-p -s TYPE -s PTTYPE -o export /dev/disk/fake-path",
+					stdout: "DEVNAME=/dev/sdb\nTYPE=%v",
+				},
+				{
+					cmd:    "fsck",
+					args:   "-a /dev/disk/fake-path",
+					stdout: "",
+				},
+				{
+					cmd:    "blockdev",
+					args:   "--getro /dev/disk/fake-path",
+					stdout: "%v",
+				},
+			},
+		},
 		{
 			name: "Invalid request (Bad Access Mode)",
 			req: &csi.NodeStageVolumeRequest{
@@ -1202,6 +1238,24 @@ func TestNodeStageVolume(t *testing.T) {
 			},
 			expErrCode: codes.InvalidArgument,
 		},
+		{
+			name: "Invalid request, block size mismatch",
+			req: &csi.NodeStageVolumeRequest{
+				VolumeId:          volumeID,
+				StagingTargetPath: stagingPath,
+				VolumeCapability:  stdVolCap,
+				PublishContext:    map[string]string{common.ContextDiskSizeGB: "10"},
+			},
+			deviceSize: 5,
+			expErrCode: codes.Internal,
+			expCommandList: []fakeCmd{
+				{
+					cmd:    "blockdev",
+					args:   "--getsize64 /dev/disk/fake-path",
+					stdout: "%v",
+				},
+			},
+		},
 	}
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {

test/remote/instance.go

@@ -248,9 +248,9 @@ func (i *InstanceInfo) CreateOrGetInstance(localSSDCount int) error {
 			i.externalIP = externalIP
 		}
 
-		if sshOut, err := i.SSHCheckAlive(); err != nil {
+		if err := i.SSHCheckAlive(); err != nil {
 			err = fmt.Errorf("Instance %v in state RUNNING but not available by SSH: %v", i.cfg.Name, err.Error())
-			klog.Warningf("SSH encountered an error: %v, output: %v", err, sshOut)
+			klog.Warningf("SSH encountered an error: %v", err)
 			return false, nil
 		}
 		klog.V(4).Infof("Instance %v in state RUNNING and available by SSH", i.cfg.Name)

test/remote/ssh.go

@@ -22,7 +22,9 @@ import (
 	"os/exec"
 	"os/user"
 	"strings"
+	"time"
 
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/klog/v2"
 )
 
@@ -88,8 +90,14 @@ func (i *InstanceInfo) SSHNoSudo(cmd ...string) (string, error) {
 }
 
 // SSHCheckAlive just pings the server quickly to check whether it is reachable by SSH
-func (i *InstanceInfo) SSHCheckAlive() (string, error) {
-	return runSSHCommand("ssh", []string{i.GetSSHTarget(), "-o", "ConnectTimeout=10", "--", "echo"}...)
+func (i *InstanceInfo) SSHCheckAlive() error {
+	return wait.Poll(5*time.Second, time.Minute, func() (bool, error) {
+		out, err := runSSHCommand("ssh", []string{i.GetSSHTarget(), "-o", "ConnectTimeout=10", "--", "echo"}...)
+		if err != nil {
+			klog.V(2).Infof("ssh error, retrying: %v, %s", err, out)
+		}
+		return err == nil, nil
+	})
 }
 
 // runSSHCommand executes the ssh or scp command, adding the flag provided --ssh-options