Update caching logic and add E2E tests

Author: Sneha-at

Dockerfile

@@ -81,6 +81,7 @@ COPY --from=debian /sbin/blkdeactivate /sbin/blkdeactivate
 COPY --from=debian /sbin/dmsetup /sbin/dmsetup
 COPY --from=debian /sbin/dmstats /sbin/dmstats
 # End of dependencies for LVM
+COPY --from=debian /sbin/mdadm /sbin/mdadm
 COPY --from=debian /sbin/mke2fs /sbin/mke2fs
 COPY --from=debian /sbin/mkfs* /sbin/
 COPY --from=debian /sbin/resize2fs /sbin/resize2fs

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

@@ -234,6 +234,10 @@ func handle() {
 		}
 	}
 
+	if *enableDataCacheFlag {
+		klog.V(2).Info("Raiding local ssd")
+		driver.RaidLocalSsds()
+	}
 	err = gceDriver.SetupGCEDriver(driverName, version, extraVolumeLabels, extraTags, identityServer, controllerServer, nodeServer)
 	if err != nil {
 		klog.Fatalf("Failed to initialize GCE CSI Driver: %v", err.Error())
@@ -251,7 +255,6 @@ func handle() {
 	gce.WaitForOpBackoff.Steps = *waitForOpBackoffSteps
 	gce.WaitForOpBackoff.Cap = *waitForOpBackoffCap
 
-	driver.RaidLocalSsds()
 	gceDriver.Run(*endpoint, *grpcLogCharCap, *enableOtelTracing)
 }
 

deploy/kubernetes/images/stable-master/image.yaml

@@ -51,6 +51,6 @@ imageTag:
   name: gke.gcr.io/gcp-compute-persistent-disk-csi-driver
   # Don't change stable image without changing pdImagePlaceholder in
   # test/k8s-integration/main.go
-  newName: gcr.io/snehaaradhey-joonix/gcp-compute-persistent-disk-csi-driver
-  newTag: "datacache"
+  newName: registry.k8s.io/cloud-provider-gcp/gcp-compute-persistent-disk-csi-driver
+  newTag: "v1.13.2"
 ---

pkg/common/constants.go

@@ -36,4 +36,10 @@ const (
 	// Data cache mode
 	DataCacheModeWriteBack    = "writeback"
 	DataCacheModeWriteThrough = "writethrough"
+
+	ContextDataCacheSize = "data-cache-size"
+	ContextDataCacheMode = "data-cache-mode"
+
+	// Keys in the publish context
+	ContexLocalSsdCacheSize = "local-ssd-cache-size"
 )

pkg/common/runcmd.go

@@ -4,8 +4,6 @@ import (
 	"fmt"
 	"os/exec"
 	"strings"
-
-	"k8s.io/klog/v2"
 )
 
 const (
@@ -16,9 +14,7 @@ const (
 // RunCommand wraps a k8s exec to deal with the no child process error. Same as exec.CombinedOutput.
 // On error, the output is included so callers don't need to echo it again.
 func RunCommand(cmd string, args ...string) ([]byte, error) {
-	klog.V(2).Infof("====== Start RunCommand ======")
 	execCmd := exec.Command(cmd, args...)
-	klog.V(2).Infof("=======Running command %v==========", execCmd)
 	output, err := execCmd.CombinedOutput()
 	if err != nil {
 		if err.Error() == errNoChildProcesses {

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

@@ -12,11 +12,14 @@ import (
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
 )
 
-const cacheSuffix = "csi-fast"
-const mainLvSuffix = "csi-main"
-const raidedLocalSsdName = "csi-driver-data-cache"
-const raidMode = "0"
-const raidedLssdPrefix = "/dev/md/"
+const (
+	cacheSuffix        = "csi-fast"
+	mainLvSuffix       = "csi-main"
+	raidedLocalSsdName = "csi-driver-data-cache"
+	raidMode           = "0"
+	raidedLssdPrefix   = "/dev/md/"
+	raidedLocalSsdPath = raidedLssdPrefix + raidedLocalSsdName
+)
 
 func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId string) (string, error) {
 	volumeId := req.GetVolumeId()
@@ -32,7 +35,6 @@ func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId str
 	if err != nil {
 		klog.Errorf("vgscan error %v: %s", err, info)
 	}
-	klog.V(2).Infof("====== vgscan info %v  ======", string(info))
 	klog.V(2).Infof("====== vgscan info contains volumeGroupName or not %v ======", strings.Contains(string(info), volumeGroupName))
 	// Check if the required volume group already exists
 	if strings.Contains(string(info), volumeGroupName) {
@@ -41,28 +43,12 @@ func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId str
 		// Clean up Volume Group before adding the PD
 		reduceVolumeGroup(volumeGroupName, true)
 	} else {
-		err := createVg(volumeGroupName, devicePath)
+		err := createVg(volumeGroupName, devicePath, raidedLocalSsdPath)
 		if err != nil {
 			return mainDevicePath, err
 		}
-		// VG doesn't exist so throw error
-		klog.Errorf("No VG on Node, ensure Volume group on node")
 	}
 
-	// Verify if the PD is already in the required volume group
-	// args = []string{
-	// 	"--select",
-	// 	"vg_name",
-	// 	volumeGroupName
-	// }
-	// vgPvs, err := common.RunCommand("pvs", args...)
-	// if err != nil {
-	// 	klog.Errorf("Errored while checking Physical Volume(PV)s for the volume group %v: %s", err, vgPvs)
-	// }
-	// if strings.Contains(string(vgPvs), devicePath) {
-	// 	klog.V(2).Infof("====Physical Volume(PV) already exists in the Volume Group=====")
-	// } else {
-
 	// Check if the Physical Volume(PV) is part of some other volume group
 	args = []string{
 		"--select",
@@ -82,7 +68,6 @@ func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId str
 	infoString = strings.ReplaceAll(infoString, ".", "")
 	infoString = strings.ReplaceAll(infoString, "\"", "")
 	infoSlice := strings.Split(strings.TrimSpace(infoString), " ")
-	klog.V(2).Info("============ infoSlice %s =============", infoSlice[(len(infoSlice)-1)], infoSlice)
 	vgNameForPv := strings.TrimSpace(infoSlice[(len(infoSlice) - 1)])
 	klog.V(2).Infof("============ Physical volume is part of Volume group: %v=======", vgNameForPv)
 	if vgNameForPv == volumeGroupName {
@@ -94,10 +79,13 @@ func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId str
 		if err != nil {
 			klog.Errorf("Errored while deactivating VG %v: err: %v: %s", vgNameForPv, err, info)
 		}
+
+		reduceVolumeGroup(vgNameForPv, false)
 		// Uncache LV
 		args = []string{
 			"--uncache",
 			vgNameForPv + "/" + mainLvName,
+			"--force",
 		}
 		info, err = common.RunCommand("lvconvert", args...)
 		if err != nil {
@@ -112,12 +100,6 @@ func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId str
 			klog.Errorf("Errored while merging Volume group %s into %s %v: %s", vgNameForPv, volumeGroupName, err, info)
 		}
 
-		klog.V(2).Infof("==========Remove VG from node %v ==========", vgNameForPv)
-		info, err = common.RunCommand("vgremove", []string{vgNameForPv, "-y"}...)
-		if err != nil {
-			klog.Errorf("Errored while removing Volume group %s: info:%s, error:%v", vgNameForPv, err, info)
-		}
-
 	} else {
 		klog.V(2).Infof("==========Extend Node VG %v for PV %v==========", volumeGroupName, devicePath)
 		info, err := common.RunCommand("vgextend", []string{volumeGroupName, devicePath}...)
@@ -135,7 +117,6 @@ func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId str
 	}
 	lvList, err := common.RunCommand("lvs", args...)
 	if err != nil {
-		klog.V(2).Infof("====== lvs error %v: %s ======", err, info)
 		return mainDevicePath, fmt.Errorf("lv list error %w: %s", err, info)
 	}
 	klog.Info("=============== Got LVs %s on Volume group %s ============", string(lvList), volumeGroupName)
@@ -153,13 +134,11 @@ func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId str
 		}
 		info, err = common.RunCommand("lvcreate", args...)
 		if err != nil {
-			klog.V(2).Infof("====== lvcreate error %v: %s ======", err, info)
 			return mainDevicePath, fmt.Errorf("lvcreate error %w: %s", err, info)
 		}
 
 	}
-	// Replace this with RAIDed Local SSD
-	cachePoolName := "dev/nvme0n1"
+	cachePoolName := raidedLocalSsdPath
 	// Verify caching is setup for PD
 	args = []string{
 		"--select",
@@ -169,17 +148,16 @@ func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId str
 	}
 	poolName, err := common.RunCommand("lvs", args...)
 	if err != nil {
-		klog.V(2).Infof("====== lvcreate error %v: %s ======", err, info)
 		return mainDevicePath, fmt.Errorf("lvcreate error %w: %s", err, info)
 	}
 	cacheLvName := getLvName(cacheSuffix, volumeId)
 	if strings.Contains(string(poolName), "csi-fast") {
 		// Validate that cache is setup for required size
 		klog.V(2).Infof("================Validate Cache is setup for correct size and mode===============")
 	} else {
-		fastCacheSize := req.GetPublishContext()[contexLocalSsdCacheSize] + "Gi"
+		fastCacheSize := req.GetPublishContext()[common.ContexLocalSsdCacheSize]
+		chunkSize := "960" // Cannot use default chunk size(64KiB) as it errors on maxChunksAllowed. Unit - KiB
 		klog.V(2).Infof("====== fastCacheSize is %v ======", fastCacheSize)
-		// lvcreate -n fast -L 50G cachegroup /dev/nvme0n1
 		klog.V(2).Infof("====== lvcreate fast cache layer again with the VolumeGroup %v======", volumeGroupName)
 		args = []string{
 			"--yes",
@@ -197,7 +175,6 @@ func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId str
 		}
 
 		// Once caching is setup, link the PD to cache
-		// lvconvert --type cache --cachevol fast --zero y --cachemode writethrough cachegroup/main --force -y
 		klog.V(2).Infof("====== lvconvert fast and main to cache ======")
 		args = []string{
 			"--type",
@@ -207,8 +184,10 @@ func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId str
 			"--zero",
 			"y",
 			"--cachemode",
-			req.GetPublishContext()[contextDataCacheMode],
+			req.GetPublishContext()[common.ContextDataCacheMode],
 			volumeGroupName + "/" + mainLvName,
+			"--chunksize",
+			string(chunkSize),
 			"--force",
 			"-y",
 		}
@@ -247,7 +226,6 @@ func cleanupCache(volumeId string, nodeId string) error {
 	}
 	info, err = common.RunCommand("lvconvert", args...)
 	if err != nil {
-		klog.Errorf("Errored while uncaching the disk  %v: %s", err, info)
 		return fmt.Errorf("errored while uncaching the disk %w: %s", err, info)
 	}
 	return nil
@@ -265,14 +243,14 @@ func getLvName(suffix string, volumeId string) string {
 	return fmt.Sprintf("%s-%s", suffix, pvcName)
 }
 
-func createVg(volumeGroupName string, devicePath string) error {
+func createVg(volumeGroupName string, devicePath string, raidedLocalSsds string) error {
 	// No existing volume group
 	klog.V(2).Infof("====== vgcreate ======")
 	args := []string{
 		"--zero",
 		"y",
 		volumeGroupName,
-		"/dev/nvme0n1",
+		raidedLocalSsds,
 	}
 	info, err := common.RunCommand("vgcreate", args...)
 	if err != nil {
@@ -307,48 +285,56 @@ func reduceVolumeGroup(volumeGroupName string, force bool) {
 }
 
 func RaidLocalSsds() error {
-	isRaided, err := isRaided()
+	isAlreadyRaided, err := isRaided()
 	if err != nil {
 		klog.V(2).Info("======Errored while scanning for available LocalSSDs err:%v; continuing Raiding=======", err)
-	} else if isRaided {
+	} else if isAlreadyRaided {
 		klog.V(2).Infof("===============Local SSDs are already RAIDed==============")
 		return nil
 	}
-	info, err := common.RunCommand("nvme", []string{"list"}...)
+	info, err := common.RunCommand("nvme", []string{"list", "-o", "json"}...)
 	if err != nil {
-		klog.Errorf("nvme list error %v: %s", err, info)
 		return fmt.Errorf("errored while scanning available NVME disks info: %v; err:%v", info, err)
 	}
-	klog.V(2).Infof("==========NVME list %v========", string(info))
 	infoString := strings.TrimSpace(strings.ReplaceAll(string(info), "\n", " "))
+	klog.V(2).Infof("=============NVME list %v ============", infoString)
 	infoString = strings.ReplaceAll(infoString, "\"", "")
-	infoSlice := strings.Split(strings.TrimSpace(infoString), " ")
+	infoString = strings.ReplaceAll(infoString, " :", ":")
+	infoString = strings.ReplaceAll(infoString, ": ", ":")
+	infoString = strings.ReplaceAll(infoString, ",", " ")
+	infoSlice := strings.Split(infoString, " ")
 	diskList := []string{}
 	for _, diskInfo := range infoSlice {
 		diskName := strings.TrimSpace(diskInfo)
-		if strings.HasPrefix(diskName, "/dev/n") {
+		if strings.Contains(diskName, "DevicePath") {
+			diskName := strings.TrimSpace(strings.Split(diskName, ":")[1])
 			diskList = append(diskList, diskName)
 		}
 	}
 	nvmeDiskCount := len(diskList)
 	nvmeDiskList := strings.Join(diskList, " ")
-	klog.V(2).Infof("========= nvmeDiskCount %v; nvmeDislList: %v ================", nvmeDiskCount, nvmeDiskList)
+	klog.V(2).Infof("========= nvmeDiskCount %v; nvmeDiskList: %v; diskList %v================", nvmeDiskCount, nvmeDiskList, diskList)
 	args := []string{
 		"--create",
 		raidedLssdPrefix + raidedLocalSsdName,
-		"-l=" + raidMode,
+		"-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
 		"--force",
 		"-n",
 		strconv.Itoa(nvmeDiskCount),
-		nvmeDiskList,
 	}
+	args = append(args, diskList...)
 	info, err = common.RunCommand("mdadm", args...)
 	if err != nil {
-		klog.Errorf("Errored while RAIDing LSSDs %v: %s", err, info)
 		return fmt.Errorf("errored while RAIDing LSSDs info: %v; err:%v", info, err)
-	} else {
-		klog.V(2).Infof("========RAIDed Local SSDs in mode 0===========")
+	}
+	// Validate if Raided successfully
+	isAlreadyRaided, err = isRaided()
+	if err != nil {
+		klog.V(2).Info("======Errored while scanning for available raided LocalSSDs err:%v=======", err)
+	}
+	if !isAlreadyRaided {
+		return fmt.Errorf("failed raiding, raided device not found on scanning")
 	}
 	return nil
 }
@@ -363,7 +349,7 @@ func isRaided() (bool, error) {
 		return false, fmt.Errorf("errored while scanning for raided LSSD %v: %s", err, info)
 	}
 	klog.V(2).Infof("=========== Got LSSDs %v===========", string(info))
-	if strings.Contains(string(info), raidedLssdPrefix+raidedLocalSsdName) {
+	if info != nil && strings.Contains(string(info), raidedLocalSsdName) {
 		return true, nil
 	}
 	return false, nil

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

@@ -43,6 +43,11 @@ import (
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/metrics"
 )
 
+var (
+	// Keys in the volume context.
+	contextForceAttach = "force-attach"
+)
+
 type GCEControllerServer struct {
 	Driver        *GCEDriver
 	CloudProvider gce.GCECompute
@@ -190,14 +195,6 @@ const (
 	// See https://github.com/grpc/grpc/blob/master/include/grpc/impl/codegen/grpc_types.h#L503)
 	maxListVolumesResponseEntries = 500
 
-	// Keys in the volume context.
-	contextForceAttach   = "force-attach"
-	contextDataCacheSize = "data-cache-size"
-	contextDataCacheMode = "data-cache-mode"
-
-	// Keys in the publish context
-	contexLocalSsdCacheSize = "local-ssd-cache-size"
-
 	resourceApiScheme  = "https"
 	resourceApiService = "compute"
 	resourceProject    = "projects"
@@ -690,10 +687,10 @@ func (gceCS *GCEControllerServer) executeControllerPublishVolume(ctx context.Con
 	klog.V(2).Infof("====== ControllerPublishVolume VolumeContext is %v ======", req.GetVolumeContext())
 	// Set data cache publish context
 	if gceCS.enableDataCache && req.GetVolumeContext() != nil {
-		if req.GetVolumeContext()[contextDataCacheSize] != "" {
+		if req.GetVolumeContext()[common.ContextDataCacheSize] != "" {
 			pubVolResp.PublishContext = map[string]string{}
-			pubVolResp.PublishContext[contexLocalSsdCacheSize] = req.GetVolumeContext()[contextDataCacheSize]
-			pubVolResp.PublishContext[contextDataCacheMode] = req.GetVolumeContext()[contextDataCacheMode]
+			pubVolResp.PublishContext[common.ContexLocalSsdCacheSize] = req.GetVolumeContext()[common.ContextDataCacheSize]
+			pubVolResp.PublishContext[common.ContextDataCacheMode] = req.GetVolumeContext()[common.ContextDataCacheMode]
 		}
 	}
 
@@ -2000,8 +1997,8 @@ func generateCreateVolumeResponse(disk *gce.CloudDisk, zones []string, params co
 		if createResp.Volume.VolumeContext == nil {
 			createResp.Volume.VolumeContext = map[string]string{}
 		}
-		createResp.Volume.VolumeContext[contextDataCacheMode] = dataCacheParams.DataCacheMode
-		createResp.Volume.VolumeContext[contextDataCacheSize] = dataCacheParams.DataCacheSize
+		createResp.Volume.VolumeContext[common.ContextDataCacheMode] = dataCacheParams.DataCacheMode
+		createResp.Volume.VolumeContext[common.ContextDataCacheSize] = dataCacheParams.DataCacheSize
 	}
 	snapshotID := disk.GetSnapshotId()
 	imageID := disk.GetImageId()