update validation logic for Data Cache to distinguish user and non-user errors when RAIDing fails

Author: hungnguyen243

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

@@ -246,9 +246,10 @@ func handle() {
 			klog.Fatalf("Failed to set up metadata service: %v", err.Error())
 		}
 		nsArgs := driver.NodeServerArgs{
-			EnableDeviceInUseCheck: *enableDeviceInUseCheck,
-			DeviceInUseTimeout:     *deviceInUseTimeout,
-			EnableDataCache:        *enableDataCacheFlag,
+			EnableDeviceInUseCheck:   *enableDeviceInUseCheck,
+			DeviceInUseTimeout:       *deviceInUseTimeout,
+			EnableDataCache:          *enableDataCacheFlag,
+			DataCacheEnabledNodePool: isDataCacheEnabledNodePool(ctx, *nodeName),
 		}
 		nodeServer = driver.NewNodeServer(gceDriver, mounter, deviceUtils, meta, statter, nsArgs)
 		if *maxConcurrentFormatAndMount > 0 {
@@ -344,6 +345,14 @@ func urlFlag(target **url.URL, name string, usage string) {
 	})
 }
 
+func isDataCacheEnabledNodePool(ctx context.Context, nodeName string) bool {
+	dataCacheLSSDCount, err := driver.GetDataCacheCountFromNodeLabel(ctx, nodeName)
+	if err != nil || dataCacheLSSDCount == 0 {
+		return false
+	}
+	return true
+}
+
 func fetchLssdsForRaiding(lssdCount int) ([]string, error) {
 	allLssds, err := driver.FetchAllLssds()
 	if err != nil {

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

@@ -225,7 +225,7 @@ func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId str
 	return mainDevicePath, nil
 }
 
-func ValidateDataCacheConfig(dataCacheMode string, dataCacheSize string, ctx context.Context, nodeName string) error {
+func ValidateDataCacheConfig(dataCacheMode string, dataCacheSize string, ctx context.Context) error {
 	if dataCacheMode != "" && dataCacheSize != "" {
 		isAlreadyRaided, err := IsRaided()
 		if err != nil {
@@ -236,8 +236,7 @@ func ValidateDataCacheConfig(dataCacheMode string, dataCacheSize string, ctx con
 		}
 		return nil
 	}
-	klog.V(4).Infof("Data Cache is not enabled for PVC (data-cache-size: %v, data-cache-mode: %v). Please set both these parameters in StorageClass to enable caching", dataCacheSize, dataCacheMode)
-	return nil
+	return fmt.Errorf("Data Cache is not enabled for PVC (data-cache-size: %v, data-cache-mode: %v). Please set both parameters in StorageClass to enable caching", dataCacheSize, dataCacheMode)
 }
 
 func GetDataCacheCountFromNodeLabel(ctx context.Context, nodeName string) (int, error) {

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

@@ -147,15 +147,16 @@ func NewIdentityServer(gceDriver *GCEDriver) *GCEIdentityServer {
 
 func NewNodeServer(gceDriver *GCEDriver, mounter *mount.SafeFormatAndMount, deviceUtils deviceutils.DeviceUtils, meta metadataservice.MetadataService, statter mountmanager.Statter, args NodeServerArgs) *GCENodeServer {
 	return &GCENodeServer{
-		Driver:                 gceDriver,
-		Mounter:                mounter,
-		DeviceUtils:            deviceUtils,
-		MetadataService:        meta,
-		volumeLocks:            common.NewVolumeLocks(),
-		VolumeStatter:          statter,
-		enableDeviceInUseCheck: args.EnableDeviceInUseCheck,
-		deviceInUseErrors:      newDeviceErrMap(args.DeviceInUseTimeout),
-		EnableDataCache:        args.EnableDataCache,
+		Driver:                   gceDriver,
+		Mounter:                  mounter,
+		DeviceUtils:              deviceUtils,
+		MetadataService:          meta,
+		volumeLocks:              common.NewVolumeLocks(),
+		VolumeStatter:            statter,
+		enableDeviceInUseCheck:   args.EnableDeviceInUseCheck,
+		deviceInUseErrors:        newDeviceErrMap(args.DeviceInUseTimeout),
+		EnableDataCache:          args.EnableDataCache,
+		DataCacheEnabledNodePool: args.DataCacheEnabledNodePool,
 	}
 }
 

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

@@ -41,12 +41,13 @@ import (
 )
 
 type GCENodeServer struct {
-	Driver          *GCEDriver
-	Mounter         *mount.SafeFormatAndMount
-	DeviceUtils     deviceutils.DeviceUtils
-	VolumeStatter   mountmanager.Statter
-	MetadataService metadataservice.MetadataService
-	EnableDataCache bool
+	Driver                   *GCEDriver
+	Mounter                  *mount.SafeFormatAndMount
+	DeviceUtils              deviceutils.DeviceUtils
+	VolumeStatter            mountmanager.Statter
+	MetadataService          metadataservice.MetadataService
+	EnableDataCache          bool
+	DataCacheEnabledNodePool bool
 
 	// A map storing all volumes with ongoing operations so that additional operations
 	// for that same volume (as defined by VolumeID) return an Aborted error
@@ -81,6 +82,8 @@ type NodeServerArgs struct {
 	DeviceInUseTimeout time.Duration
 
 	EnableDataCache bool
+
+	DataCacheEnabledNodePool bool
 }
 
 var _ csi.NodeServer = &GCENodeServer{}
@@ -337,17 +340,20 @@ func (ns *GCENodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStage
 
 	klog.Infof("Successfully found attached GCE PD %q at device path %s.", volumeKey.Name, devicePath)
 
-	if ns.EnableDataCache && req.GetPublishContext()[common.ContextDataCacheSize] != "" {
+	if ns.EnableDataCache && (req.GetPublishContext()[common.ContextDataCacheSize] != "" || req.GetPublishContext()[common.ContextDataCacheMode] != "") {
 		if len(nodeId) == 0 {
 			return nil, status.Error(codes.InvalidArgument, "NodeStageVolume Node ID must be provided")
 		}
 		devFsPath, err := filepath.EvalSymlinks(devicePath)
 		if err != nil {
 			klog.Errorf("filepath.EvalSymlinks(%q) failed when trying to create volume group: %v", devicePath, err)
 		}
-		configError := ValidateDataCacheConfig(req.GetPublishContext()[common.ContextDataCacheMode], req.GetPublishContext()[common.ContextDataCacheSize], ctx, nodeId)
+		configError := ValidateDataCacheConfig(req.GetPublishContext()[common.ContextDataCacheMode], req.GetPublishContext()[common.ContextDataCacheSize], ctx)
 		if configError != nil {
-			return nil, status.Error(codes.Internal, fmt.Sprintf("Error validate configuration for Data Cache: %v", err.Error()))
+			if ns.DataCacheEnabledNodePool {
+				return nil, status.Error(codes.DataLoss, fmt.Sprintf("Error validate configuration for Data Cache: %v", configError.Error()))
+			}
+			return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("Data Cache PVC is requested for an incompatible node pool: %v", configError.Error()))
 		}
 		devicePath, err = setupCaching(devFsPath, req, nodeId)
 		if err != nil {

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

@@ -52,7 +52,7 @@ func getTestGCEDriverWithCustomMounter(t *testing.T, mounter *mount.SafeFormatAn
 func getCustomTestGCEDriver(t *testing.T, mounter *mount.SafeFormatAndMount, deviceUtils deviceutils.DeviceUtils, metaService metadataservice.MetadataService) *GCEDriver {
 	gceDriver := GetGCEDriver()
 	enableDataCache := false
-	nodeServer := NewNodeServer(gceDriver, mounter, deviceUtils, metaService, mountmanager.NewFakeStatter(mounter), NodeServerArgs{true, 0, enableDataCache})
+	nodeServer := NewNodeServer(gceDriver, mounter, deviceUtils, metaService, mountmanager.NewFakeStatter(mounter), NodeServerArgs{true, 0, enableDataCache, false /*dataCacheEnableNodePool */})
 	err := gceDriver.SetupGCEDriver(driver, "test-vendor", nil, nil, nil, nil, nodeServer)
 	if err != nil {
 		t.Fatalf("Failed to setup GCE Driver: %v", err)
@@ -63,7 +63,7 @@ func getCustomTestGCEDriver(t *testing.T, mounter *mount.SafeFormatAndMount, dev
 func getTestBlockingMountGCEDriver(t *testing.T, readyToExecute chan chan struct{}) *GCEDriver {
 	gceDriver := GetGCEDriver()
 	mounter := mountmanager.NewFakeSafeBlockingMounter(readyToExecute)
-	nodeServer := NewNodeServer(gceDriver, mounter, deviceutils.NewFakeDeviceUtils(false), metadataservice.NewFakeService(), mountmanager.NewFakeStatter(mounter), NodeServerArgs{true, 0, true})
+	nodeServer := NewNodeServer(gceDriver, mounter, deviceutils.NewFakeDeviceUtils(false), metadataservice.NewFakeService(), mountmanager.NewFakeStatter(mounter), NodeServerArgs{true, 0, true, false /*dataCacheEnableNodePool */})
 	err := gceDriver.SetupGCEDriver(driver, "test-vendor", nil, nil, nil, nil, nodeServer)
 	if err != nil {
 		t.Fatalf("Failed to setup GCE Driver: %v", err)
@@ -75,7 +75,7 @@ func getTestBlockingFormatAndMountGCEDriver(t *testing.T, readyToExecute chan ch
 	gceDriver := GetGCEDriver()
 	enableDataCache := true
 	mounter := mountmanager.NewFakeSafeBlockingMounter(readyToExecute)
-	nodeServer := NewNodeServer(gceDriver, mounter, deviceutils.NewFakeDeviceUtils(false), metadataservice.NewFakeService(), mountmanager.NewFakeStatter(mounter), NodeServerArgs{true, 0, enableDataCache}).WithSerializedFormatAndMount(5*time.Second, 1)
+	nodeServer := NewNodeServer(gceDriver, mounter, deviceutils.NewFakeDeviceUtils(false), metadataservice.NewFakeService(), mountmanager.NewFakeStatter(mounter), NodeServerArgs{true, 0, enableDataCache, false /*dataCacheEnableNodePool */}).WithSerializedFormatAndMount(5*time.Second, 1)
 
 	err := gceDriver.SetupGCEDriver(driver, "test-vendor", nil, nil, nil, nil, nodeServer)
 	if err != nil {