Merge pull request #1719 from Sneha-at/data-cache

Add separate flags for node and controller and hash nodeId

Author: mattcary

Dockerfile

@@ -148,6 +148,8 @@ SHELL ["/bin/bash", "-c"]
 RUN /bin/sed -i -e "s/.*allow_mixed_block_sizes = 0.*/	allow_mixed_block_sizes = 1/" /etc/lvm/lvm.conf
 RUN /bin/sed -i -e "s/.*udev_sync = 1.*/	udev_sync = 0/" /etc/lvm/lvm.conf
 RUN /bin/sed -i -e "s/.*udev_rules = 1.*/	udev_rules = 0/" /etc/lvm/lvm.conf
+RUN /bin/sed -i -e "s/.*locking_dir = .*/	 locking_dir = \"\/tmp\"/" /etc/lvm/lvm.conf
+
 
 # Build stage used for validation of the output-image
 # See validate-container-linux-* targets in Makefile

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

@@ -67,11 +67,12 @@ var (
 	maxConcurrentFormat     = flag.Int("max-concurrent-format", 1, "The maximum number of concurrent format exec calls")
 	concurrentFormatTimeout = flag.Duration("concurrent-format-timeout", 1*time.Minute, "The maximum duration of a format operation before its concurrency token is released")
 
-	maxConcurrentFormatAndMount = flag.Int("max-concurrent-format-and-mount", 1, "If set then format and mount operations are serialized on each node. This is stronger than max-concurrent-format as it includes fsck and other mount operations")
-	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")
+	maxConcurrentFormatAndMount   = flag.Int("max-concurrent-format-and-mount", 1, "If set then format and mount operations are serialized on each node. This is stronger than max-concurrent-format as it includes fsck and other mount operations")
+	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")
+	enableControllerDataCacheFlag = flag.Bool("enable-controller-data-cache", false, "If set to true, the CSI Driver will allow volumes to be provisioned with data cache configuration")
+	enableNodeDataCacheFlag       = flag.Bool("enable-node-data-cache", false, "If set to true, the CSI Driver will allow volumes to be provisioned with data cache configuration")
 
 	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")
 	multiZoneVolumeHandleEnableFlag    = flag.Bool("multi-zone-volume-handle-enable", false, "If set to true, the multi-zone volumeHandle feature will be enabled")
@@ -209,7 +210,7 @@ func handle() {
 		}
 		initialBackoffDuration := time.Duration(*errorBackoffInitialDurationMs) * time.Millisecond
 		maxBackoffDuration := time.Duration(*errorBackoffMaxDurationMs) * time.Millisecond
-		controllerServer = driver.NewControllerServer(gceDriver, cloudProvider, initialBackoffDuration, maxBackoffDuration, fallbackRequisiteZones, *enableStoragePoolsFlag, *enableDataCacheFlag, multiZoneVolumeHandleConfig, listVolumesConfig)
+		controllerServer = driver.NewControllerServer(gceDriver, cloudProvider, initialBackoffDuration, maxBackoffDuration, fallbackRequisiteZones, *enableStoragePoolsFlag, *enableControllerDataCacheFlag, multiZoneVolumeHandleConfig, listVolumesConfig)
 	} else if *cloudConfigFilePath != "" {
 		klog.Warningf("controller service is disabled but cloud config given - it has no effect")
 	}
@@ -227,13 +228,13 @@ func handle() {
 		if err != nil {
 			klog.Fatalf("Failed to set up metadata service: %v", err.Error())
 		}
-		nodeServer = driver.NewNodeServer(gceDriver, mounter, deviceUtils, meta, statter, *enableDataCacheFlag)
+		nodeServer = driver.NewNodeServer(gceDriver, mounter, deviceUtils, meta, statter, *enableNodeDataCacheFlag)
 		if *maxConcurrentFormatAndMount > 0 {
 			nodeServer = nodeServer.WithSerializedFormatAndMount(*formatAndMountTimeout, *maxConcurrentFormatAndMount)
 		}
 	}
 
-	if *enableDataCacheFlag {
+	if *enableNodeDataCacheFlag {
 		klog.V(2).Info("Raiding local ssds to setup data cache")
 		err := driver.RaidLocalSsds()
 		if err != nil {

pkg/common/utils.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"hash/fnv"
 	"net/http"
 	"regexp"
 	"strings"
@@ -72,6 +73,7 @@ const (
 	// Full or partial URL of the machine type resource, in the format:
 	//   zones/zone/machineTypes/machine-type
 	machineTypePattern = "zones/[^/]+/machineTypes/([^/]+)$"
+	alphanums          = "bcdfghjklmnpqrstvwxz2456789"
 )
 
 var (
@@ -617,3 +619,16 @@ func NewLimiter(limit, burst int, emptyBucket bool) *rate.Limiter {
 
 	return limiter
 }
+
+// shortString is inspired by k8s.io/apimachinery/pkg/util/rand.SafeEncodeString, but takes data from a hash.
+func ShortString(s string) string {
+	hasher := fnv.New128a()
+	hasher.Write([]byte(s))
+	sum := hasher.Sum([]byte{})
+	const sz = 8
+	short := make([]byte, sz)
+	for i := 0; i < sz; i++ {
+		short[i] = alphanums[int(sum[i])%len(alphanums)]
+	}
+	return string(short)
+}

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

@@ -241,7 +241,8 @@ func cleanupCache(volumeId string, nodeId string) error {
 func getVolumeGroupName(nodePath string) string {
 	nodeSlice := strings.Split(nodePath, "/")
 	nodeId := nodeSlice[len(nodeSlice)-1]
-	return fmt.Sprintf("csi-vg-%s", nodeId)
+	nodeHash := common.ShortString(nodeId)
+	return fmt.Sprintf("csi-vg-%s", nodeHash)
 }
 
 func getLvName(suffix string, volumeId string) string {
@@ -257,12 +258,14 @@ func createVg(volumeGroupName string, devicePath string, raidedLocalSsds string)
 		"y",
 		volumeGroupName,
 		raidedLocalSsds,
+		"-v",
 	}
 	info, err := common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "vgcreate", args...)
 	if err != nil {
 		klog.Errorf("vgcreate error %v: %s", err, info)
 		return fmt.Errorf("vgcreate error %w: %s", err, info)
 	}
+	klog.Infof("Volume group creation succeeded for %v", volumeGroupName)
 
 	klog.V(2).Infof("============================== vgscan after vgcreate ==============================")
 	args = []string{}
@@ -298,19 +301,27 @@ func RaidLocalSsds() error {
 		klog.V(2).Infof("============================== Local SSDs are already RAIDed ==============================")
 		return nil
 	}
-	info, err := common.RunCommand("grep" /* pipedCmd */, "DevicePath" /* pipeCmdArg */, "nvme", []string{"list", "-o", "json"}...)
+	info, err := common.RunCommand("" /* pipedCmd */, "" /* pipeCmdArg */, "nvme", []string{"list", "-o", "json"}...)
 	if err != nil {
 		return fmt.Errorf("errored while scanning available NVME disks info: %v; err:%v", info, err)
 	}
-	infoString := strings.ReplaceAll(string(info), "\"", "")
-	infoString = strings.TrimSpace(strings.ReplaceAll(infoString, ",", " "))
-	infoSlice := strings.Split(infoString, "\n")
-	klog.V(2).Infof("============================== NVME list %v ==============================", infoSlice)
+	infoString := strings.TrimSpace(strings.ReplaceAll(string(info), "\n", " "))
+	klog.V(2).Infof("============================== NVME list %v ==============================", infoString)
+	infoString = strings.ReplaceAll(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)
-		diskName = strings.TrimSpace(strings.Split(diskName, ":")[1])
-		diskList = append(diskList, diskName)
+
+		if strings.Contains(diskName, "DevicePath") {
+			diskName := strings.TrimSpace(strings.Split(diskName, ":")[1])
+
+			diskList = append(diskList, diskName)
+		}
 	}
 	nvmeDiskCount := len(diskList)
 	nvmeDiskList := strings.Join(diskList, " ")

test/e2e/utils/utils.go

@@ -60,7 +60,8 @@ func GCEClientAndDriverSetup(instance *remote.InstanceInfo, computeEndpoint stri
 		"--multi-zone-volume-handle-disk-types=pd-standard",
 		"--use-instance-api-to-poll-attachment-disk-types=pd-ssd",
 		"--use-instance-api-to-list-volumes-published-nodes",
-		"--enable-data-cache",
+		"--enable-controller-data-cache",
+		"--enable-node-data-cache",
 	}
 	extra_flags = append(extra_flags, fmt.Sprintf("--compute-endpoint=%s", computeEndpoint))
 

test/run-e2e-local.sh

@@ -16,4 +16,4 @@ if hostname | grep -q c.googlers.com ; then
   CLOUDTOP_HOST=--cloudtop-host
 fi
 
-ginkgo --v --focus "Should create->attach->setup caching->write->detach->attach to different node->mount->read" "test/e2e/tests" -- --project "${PROJECT}" --service-account "${IAM_NAME}" "${CLOUDTOP_HOST}" --v=6 --logtostderr
+ginkgo --v "test/e2e/tests" -- --project "${PROJECT}" --service-account "${IAM_NAME}" "${CLOUDTOP_HOST}" --v=6 --logtostderr