update test setup to avoid running Datacache setup on machines not supporting LSSDs

Author: hungnguyen243

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

@@ -71,7 +71,7 @@ var (
 	formatAndMountTimeout       = flag.Duration("format-and-mount-timeout", 1*time.Minute, "The maximum duration of a format and mount operation before another such operation will be started. Used only if --serialize-format-and-mount")
 	fallbackRequisiteZonesFlag  = flag.String("fallback-requisite-zones", "", "Comma separated list of requisite zones that will be used if there are not sufficient zones present in requisite topologies when provisioning a disk")
 	enableStoragePoolsFlag      = flag.Bool("enable-storage-pools", false, "If set to true, the CSI Driver will allow volumes to be provisioned in Storage Pools")
-	enableDataCacheFlag         = flag.Bool("enable-data-cache", false, "If set to true, the CSI Driver will allow volumes to be provisioned with data cache configuration")
+	enableDataCacheFlag         = flag.Bool("enable-data-cache", false, "If set to true, the CSI Driver will allow volumes to be provisioned with Data Cache configuration")
 	nodeName                    = flag.String("node-name", "", "The node this driver is running on")
 
 	multiZoneVolumeHandleDiskTypesFlag = flag.String("multi-zone-volume-handle-disk-types", "", "Comma separated list of allowed disk types that can use the multi-zone volumeHandle. Used only if --multi-zone-volume-handle-enable")
@@ -123,7 +123,7 @@ func handle() {
 	if version == "" {
 		klog.Fatalf("version must be set at compile time")
 	}
-	klog.V(2).Infof("Driver vendor version %v", version)
+	klog.V(4).Infof("Driver vendor version %v", version)
 
 	// Start tracing as soon as possible
 	if *enableOtelTracing {
@@ -235,14 +235,14 @@ func handle() {
 		if *maxConcurrentFormatAndMount > 0 {
 			nodeServer = nodeServer.WithSerializedFormatAndMount(*formatAndMountTimeout, *maxConcurrentFormatAndMount)
 		}
-		if *enableDataCacheFlag {
-			if nodeName == nil || *nodeName == "" {
-				klog.Errorf("Data Cache enabled, but --node-name not passed")
-			}
-			if err := setupDataCache(ctx, *nodeName, nodeServer.MetadataService.GetName()); err != nil {
-				klog.Errorf("DataCache setup failed: %v", err)
-			}
-			go driver.StartWatcher(*nodeName)
+	}
+
+	if *enableDataCacheFlag {
+		if nodeName == nil || *nodeName == "" {
+			klog.Errorf("Data Cache enabled, but --node-name not passed")
+		}
+		if err := setupDataCache(ctx, *nodeName); err != nil {
+			klog.Errorf("Data Cache setup failed: %v", err)
 		}
 	}
 
@@ -351,7 +351,7 @@ func fetchLssdsForRaiding(lssdCount int) ([]string, error) {
 		return nil, fmt.Errorf("Error listing LSSDs with empty mountpoint: %v", err)
 	}
 
-	// We need to ensure the disks to be used for Datacache are both unRAIDed & not containing mountpoints for ephemeral storage already
+	// We need to ensure the disks to be used for Data Cache are both unRAIDed & not containing mountpoints for ephemeral storage already
 	availableLssds := slices.Filter(nil, unRaidedLssds, func(e string) bool {
 		return slices.Contains(LSSDsWithEmptyMountPoint, e)
 	})
@@ -369,36 +369,31 @@ func fetchLssdsForRaiding(lssdCount int) ([]string, error) {
 func setupDataCache(ctx context.Context, nodeName string) error {
 	isAlreadyRaided, err := driver.IsRaided()
 	if err != nil {
-		klog.V(2).Infof("Errored while scanning for available LocalSSDs err:%v; continuing Raiding", err)
+		klog.V(4).Infof("Errored while scanning for available LocalSSDs err:%v; continuing Raiding", err)
 	} else if isAlreadyRaided {
-		klog.V(2).Infof("Local SSDs are already RAIDed. Skipping Datacache setup.")
+		klog.V(4).Infof("Local SSDs are already RAIDed. Skipping Data Cache setup.")
 		return nil
 	}
 
 	lssdCount := common.LocalSSDCountForDataCache
 	if nodeName != common.TestNode {
 		var err error
 		lssdCount, err = driver.GetDataCacheCountFromNodeLabel(ctx, nodeName)
-		if lssdCount == 0 {
-			klog.Infof("Datacache is not enabled on node %v", nodeName)
-			return nil
-		}
 		if err != nil {
 			return err
 		}
+		if lssdCount == 0 {
+			klog.V(4).Infof("Data Cache is not enabled on node %v, so skipping caching setup", nodeName)
+			return nil
+		}
 	}
 	lssdNames, err := fetchLssdsForRaiding(lssdCount)
 	if err != nil {
-		klog.Fatalf("Failed to get sufficient SSDs for Datacache's caching setup: %v", err)
+		klog.Fatalf("Failed to get sufficient SSDs for Data Cache's caching setup: %v", err)
 	}
-	klog.V(2).Infof("Raiding local ssds to setup data cache: %v", lssdNames)
+	klog.V(4).Infof("Raiding local ssds to setup Data Cache: %v", lssdNames)
 	if err := driver.RaidLocalSsds(lssdNames); err != nil {
-		return fmt.Errorf("Failed to Raid local SSDs, unable to setup data caching, got error %v", err)
-	}
-
-	// Initializing data cache node (VG checks w/ raided lssd)
-	if err := driver.InitializeDataCacheNode(nodeId); err != nil {
-		return err
+		return fmt.Errorf("Failed to Raid local SSDs, unable to setup Data Cache, got error %v", err)
 	}
 
 	klog.V(4).Infof("LSSD caching is setup for the Data Cache enabled node %s", nodeName)

pkg/common/constants.go

@@ -48,7 +48,7 @@ const (
 	// Default LSSD count for datacache E2E tests
 	LocalSSDCountForDataCache = 2
 
-	// Node label for datacache
+	// Node label for Data Cache (only applicable to GKE nodes)
 	NodeLabelPrefix         = "cloud.google.com/%s"
 	DataCacheLssdCountLabel = "gke-data-cache-disk"
 )

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

@@ -16,11 +16,10 @@ import (
 )
 
 const (
-	cacheSuffix               = "csi-fast"
-	mainLvSuffix              = "csi-main"
-	raidedLocalSsdName        = "csi-driver-data-cache"
-	raidMode                  = "0"
-	initialRaidedLocalSsdPath = "/dev/md0"
+	cacheSuffix        = "csi-fast"
+	mainLvSuffix       = "csi-main"
+	raidedLocalSsdName = "csi-driver-data-cache"
+	raidMode           = "0"
 )
 
 func fetchRAIDedLocalSsdPath() (string, error) {
@@ -30,12 +29,13 @@ func fetchRAIDedLocalSsdPath() (string, error) {
 	}
 	info, err := common.RunCommand("grep", []string{raidedLocalSsdName}, "mdadm", args...)
 	if err != nil || len(info) == 0 {
-		return "", fmt.Errorf("Error getting RAIDed device path for Datacache %v, output:%v ===============", err, string(info))
+		return "", fmt.Errorf("Error getting RAIDed device path for Data Cache %v, output:%v", err, string(info))
 	}
 	infoString := strings.TrimSpace(string(info))
 	infoSlice := strings.Split(infoString, " ")
 
-	// We want to get the second element in the array, which is the path to the RAIDed device
+	// We want to get the second element in the array (sample: ARRAY /dev/md126 metadata=1.2 name=csi-driver-data-cache UUID=*),
+	//  which is the path to the RAIDed device
 	return infoSlice[1], nil
 }
 
@@ -51,7 +51,7 @@ func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId str
 	volumeGroupName := getVolumeGroupName(nodeId)
 	mainDevicePath := "/dev/" + volumeGroupName + "/" + getLvName(mainLvSuffix, volumeId)
 	mainLvName := getLvName(mainLvSuffix, volumeId)
-	klog.V(2).Infof("Volume group available on node %v ", volumeGroupName)
+	klog.V(4).Infof("Volume group available on node %v ", volumeGroupName)
 	vgExists := checkVgExists(volumeGroupName)
 	if vgExists {
 		// Clean up Volume Group before adding the PD
@@ -82,9 +82,9 @@ func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId str
 	infoString = strings.ReplaceAll(infoString, "\"", "")
 	infoSlice := strings.Split(strings.TrimSpace(infoString), " ")
 	vgNameForPv := strings.TrimSpace(infoSlice[(len(infoSlice) - 1)])
-	klog.V(2).Infof("============================== Physical volume is part of Volume group: %v ==============================", vgNameForPv)
+	klog.V(4).Infof("Physical volume is part of Volume group: %v", vgNameForPv)
 	if vgNameForPv == volumeGroupName {
-		klog.V(2).Infof("============================== Physical Volume(PV) already exists in the Volume Group ==============================")
+		klog.V(4).Infof("Physical Volume(PV) already exists in the Volume Group")
 	} else if vgNameForPv != "VG" && vgNameForPv != "" {
 
 		info, err = common.RunCommand("" /* pipedCmd */, nil /* pipedCmdArg */, "vgchange", []string{"-an", vgNameForPv}...)
@@ -157,7 +157,7 @@ func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId str
 	cacheLvName := getLvName(cacheSuffix, volumeId)
 	if isCached {
 		// Validate that cache is setup for required size
-		klog.V(2).Infof("Assuming valid data cache size and mode, resizing cache is not supported")
+		klog.V(4).Infof("Assuming valid data cache size and mode, resizing cache is not supported")
 	} else {
 		fastCacheSize := req.GetPublishContext()[common.ContextDataCacheSize]
 		chunkSize := "960" // Cannot use default chunk size(64KiB) as it errors on maxChunksAllowed. Unit - KiB
@@ -207,8 +207,8 @@ func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId str
 	return mainDevicePath, nil
 }
 
-func ValidateDataCacheConfig(dataCacheMode string, datacacheSize string, ctx context.Context, nodeName string) error {
-	if dataCacheMode != "" && datacacheSize != "" {
+func ValidateDataCacheConfig(dataCacheMode string, dataCacheSize string, ctx context.Context, nodeName string) error {
+	if dataCacheMode != "" && dataCacheSize != "" {
 		isAlreadyRaided, err := IsRaided()
 		if err != nil {
 			return fmt.Errorf("Local SSDs are not setup for caching; got error: %v", err)
@@ -218,48 +218,50 @@ func ValidateDataCacheConfig(dataCacheMode string, datacacheSize string, ctx con
 		}
 		return nil
 	}
-	klog.Infof("Data cache is not enabled for PVC")
+	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
 }
 
 func GetDataCacheCountFromNodeLabel(ctx context.Context, nodeName string) (int, error) {
-	if nodeName == common.TestNode {
-		return common.LocalSSDCountForDataCache, nil
-	}
 	cfg, err := rest.InClusterConfig()
 	// We want to capture API errors with node label fetching, so return -1
 	// in those cases instead of 0.
 	if err != nil {
-		return -1, err
+		return 0, err
 	}
 	kubeClient, err := kubernetes.NewForConfig(cfg)
 	if err != nil {
-		return -1, err
+		return 0, err
 	}
 	node, err := kubeClient.CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})
 	if err != nil {
 		// We could retry, but this error will also crashloop the driver which may be as good a way to retry as any.
-		return -1, err
+		return 0, err
 	}
 	if val, found := node.GetLabels()[fmt.Sprintf(common.NodeLabelPrefix, common.DataCacheLssdCountLabel)]; found {
 		dataCacheCount, err := strconv.Atoi(val)
 		if err != nil {
-			return -1, fmt.Errorf("Error getting Datacache's LSSD count from node label: %v", err)
+			return 0, fmt.Errorf("Error getting Data Cache's LSSD count from node label: %v", err)
 		}
-		klog.Infof("Number of local SSDs requested for Datacache: %v", dataCacheCount)
+		klog.V(4).Infof("Number of local SSDs requested for Data Cache: %v", dataCacheCount)
 		return dataCacheCount, nil
 	}
-	return 0, fmt.Errorf("Cannot get Datacache's LSSD count from node label")
+	// This will be returned for a non-Data-Cache node pool
+	return 0, nil
 }
 
 func FetchRaidedLssdCountForDatacache() (int, error) {
+	raidedPath, err := fetchRAIDedLocalSsdPath()
+	if err != nil {
+		return 0, err
+	}
 	args := []string{
 		"--detail",
-		initialRaidedLocalSsdPath,
+		raidedPath,
 	}
 	info, err := common.RunCommand("grep", []string{"Raid Devices"}, "mdadm", args...)
 	if err != nil {
-		return 0, fmt.Errorf("Error getting RAIDed devices for Datacache")
+		return 0, fmt.Errorf("Error getting RAIDed devices for Data Cache")
 	}
 	if len(info) != 0 {
 		raidedDeviceInfo := strings.Split(strings.TrimSpace(string(info)), ":")
@@ -294,7 +296,7 @@ func FetchRaidedLssds() ([]string, error) {
 		}
 	}
 
-	klog.V(2).Infof("Raided NVME list %v", raidedLssdList)
+	klog.V(4).Infof("Raided NVME list %v", raidedLssdList)
 
 	return raidedLssdList, nil
 }
@@ -309,7 +311,7 @@ func FetchAllLssds() ([]string, error) {
 	infoList := strings.Split(strings.TrimSpace(string(info)), "\n")
 	re, err := regexp.Compile("nvme_card([0-9]+)?$")
 	if err != nil {
-		klog.V(2).ErrorS(err, "Errored while compiling to check PD or LSSD")
+		klog.V(4).ErrorS(err, "Errored while compiling to check PD or LSSD")
 	}
 	for _, ssd := range infoList {
 		ssd = strings.TrimSpace(ssd)
@@ -322,7 +324,7 @@ func FetchAllLssds() ([]string, error) {
 		}
 	}
 
-	klog.V(2).Infof("NVME list %v", diskList)
+	klog.V(4).Infof("NVME list %v", diskList)
 
 	return diskList, nil
 }
@@ -358,6 +360,7 @@ func cleanupCache(volumeId string, nodeId string) error {
 		// If volume group doesn't exist then there's nothing to uncache
 		return nil
 	}
+	reduceVolumeGroup(volumeGroupName, true)
 	mainLvName := getLvName(mainLvSuffix, volumeId)
 	args := []string{
 		"-an",
@@ -404,7 +407,7 @@ func createVg(volumeGroupName string, raidedLocalSsds string) error {
 	if err != nil {
 		return fmt.Errorf("Volume group creation failed %w: %s", err, info)
 	}
-	klog.Infof("Volume group creation succeeded for %v", volumeGroupName)
+	klog.V(4).Infof("Volume group creation succeeded for %v", volumeGroupName)
 
 	args = []string{}
 	info, err = common.RunCommand("" /* pipedCmd */, nil /* pipedCmdArg */, "vgscan", args...)
@@ -431,8 +434,6 @@ func reduceVolumeGroup(volumeGroupName string, force bool) {
 func RaidLocalSsds(availableLssds []string) error {
 	args := []string{
 		"--create",
-		initialRaidedLocalSsdPath,
-		"--name",
 		raidedLocalSsdName,
 		"-l" + raidMode,
 		// Force RAIDing as sometime it might fail for caution if there is just 1 LSSD present as 1 LSSD need not be RAIDed
@@ -448,7 +449,7 @@ func RaidLocalSsds(availableLssds []string) error {
 	// Validate if Raided successfully
 	isAlreadyRaided, err := IsRaided()
 	if err != nil {
-		klog.V(2).Infof("Errored while scanning for available raided LocalSSDs err:%v=", err)
+		klog.V(4).Infof("Errored while scanning for available raided LocalSSDs err:%v=", err)
 	}
 	if !isAlreadyRaided {
 		return fmt.Errorf("failed raiding, raided device not found on scanning")

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

@@ -338,7 +338,7 @@ func (ns *GCENodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStage
 		}
 		configError := ValidateDataCacheConfig(req.GetPublishContext()[common.ContextDataCacheMode], req.GetPublishContext()[common.ContextDataCacheSize], ctx, nodeId)
 		if configError != nil {
-			return nil, status.Error(codes.Internal, fmt.Sprintf("Error validate configuration for Datacache: %v", err.Error()))
+			return nil, status.Error(codes.Internal, fmt.Sprintf("Error validate configuration for Data Cache: %v", err.Error()))
 		}
 		devicePath, err = setupCaching(devFsPath, req, nodeId)
 		if err != nil {

test/e2e/utils/utils.go

@@ -67,8 +67,11 @@ func GCEClientAndDriverSetup(instance *remote.InstanceInfo, driverConfig DriverC
 		"--use-instance-api-to-poll-attachment-disk-types=pd-ssd",
 		"--use-instance-api-to-list-volumes-published-nodes",
 		fmt.Sprintf("--fallback-requisite-zones=%s", strings.Join(driverConfig.Zones, ",")),
-		"--enable-data-cache",
-		fmt.Sprintf("--node-name=%s", utilcommon.TestNode),
+	}
+
+	if instance.GetLocalSSD() > 0 {
+		extra_flags = append(extra_flags, "--enable-data-cache")
+		extra_flags = append(extra_flags, fmt.Sprintf("--node-name=%s", utilcommon.TestNode))
 	}
 	extra_flags = append(extra_flags, fmt.Sprintf("--compute-endpoint=%s", driverConfig.ComputeEndpoint))
 	extra_flags = append(extra_flags, driverConfig.ExtraFlags...)

test/remote/instance.go

@@ -80,6 +80,10 @@ func (i *InstanceInfo) GetNodeID() string {
 	return common.CreateNodeID(i.cfg.Project, i.cfg.Zone, i.cfg.Name)
 }
 
+func (i *InstanceInfo) GetLocalSSD() int64 {
+	return i.cfg.LocalSSDCount
+}
+
 func machineTypeMismatch(curInst *compute.Instance, newInst *compute.Instance) bool {
 	if !strings.Contains(curInst.MachineType, newInst.MachineType) {
 		klog.Infof("Machine type mismatch")