Implementing watcher & reboot stability for data cache to master branch.

Author: halimsam

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

@@ -105,8 +105,12 @@ const (
 	driverName          = "pd.csi.storage.gke.io"
 	dataCacheLabel      = "datacache-storage-gke-io"
 	dataCacheLabelValue = "enabled"
+	raidedLocalSsdName  = "csi-driver-data-cache"
+	raidedLssdPrefix    = "/dev/md"
 )
 
+var raidedLocalSsdPath = raidedLssdPrefix + raidedLocalSsdName
+
 func init() {
 	// klog verbosity guide for this package
 	// Use V(2) for one time config information
@@ -266,9 +270,10 @@ func handle() {
 		if nodeName == nil || *nodeName == "" {
 			klog.Errorf("Data cache enabled, but --node-name not passed")
 		}
-		if err := setupDataCache(ctx, *nodeName); err != nil {
+		if err := setupDataCache(ctx, *nodeName, nodeServer.MetadataService.GetName()); err != nil {
 			klog.Errorf("DataCache setup failed: %v", err)
 		}
+		go driver.StartWatcher(*nodeName)
 	}
 
 	err = gceDriver.SetupGCEDriver(driverName, version, extraVolumeLabels, extraTags, identityServer, controllerServer, nodeServer)
@@ -350,7 +355,7 @@ func urlFlag(target **url.URL, name string, usage string) {
 	})
 }
 
-func setupDataCache(ctx context.Context, nodeName string) error {
+func setupDataCache(ctx context.Context, nodeName string, nodeId string) error {
 	klog.V(2).Infof("Setting up data cache for node %s", nodeName)
 	if nodeName != common.TestNode {
 		cfg, err := rest.InClusterConfig()
@@ -373,7 +378,39 @@ func setupDataCache(ctx context.Context, nodeName string) error {
 	}
 	klog.V(2).Info("Raiding local ssds to setup data cache")
 	if err := driver.RaidLocalSsds(); err != nil {
-		return fmt.Errorf("Failed to Raid local SSDs, unable to setup data caching, got error %v", err)
+		return fmt.Errorf("failed to Raid local SSDs, unable to setup data caching, got error %v", err)
+	}
+
+	// vgcreate with the raided local ssds.
+	info, err := common.RunCommand("grep", raidedLocalSsdName, "ls", raidedLssdPrefix)
+	if err != nil {
+		return fmt.Errorf("failed while listing raided devices, err: %v, output: %v", err, info)
+	}
+	infoString := strings.TrimSpace(string(info))
+	klog.V(2).Infof("Got Raided LSSD name %v", infoString)
+	raidedLocalSsdPath = raidedLssdPrefix + infoString
+	volumeGroupName := driver.GetVolumeGroupName(nodeId)
+
+	klog.V(2).Infof("vgscan before vgcreate")
+	vgExists := driver.CheckVgExists(volumeGroupName)
+	klog.V(2).Infof("vgscan info contains volumeGroupName or not %v", vgExists)
+	// Check if the required volume group already exists
+	if vgExists {
+		klog.V(2).Infof("VG exists, now check if PD is part of VG")
+
+		// Clean up Volume Group before adding the PD
+		driver.ReduceVolumeGroup(volumeGroupName, true)
+
+		// validate that raidedLSSD is part of VG
+		err = driver.ValidateRaidedLSSDinVG(volumeGroupName)
+		if err != nil {
+			return fmt.Errorf("validateRaidedLSSDinVG error: %v", err)
+		}
+	} else {
+		err := driver.CreateVg(volumeGroupName, raidedLocalSsdPath)
+		if err != nil {
+			return err
+		}
 	}
 
 	klog.V(2).Infof("Datacache enabled for node %s", nodeName)

go.mod

@@ -58,7 +58,7 @@ require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/emicklei/go-restful v2.9.5+incompatible // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
-	github.com/fsnotify/fsnotify v1.5.4 // indirect
+	github.com/fsnotify/fsnotify v1.8.0 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-openapi/jsonpointer v0.20.0 // indirect

go.sum

@@ -1032,6 +1032,8 @@ github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4
 github.com/fsnotify/fsnotify v1.5.1/go.mod h1:T3375wBYaZdLLcVNkcVbzGHY7f1l/uK5T5Ai1i3InKU=
 github.com/fsnotify/fsnotify v1.5.4 h1:jRbGcIw6P2Meqdwuo0H1p6JVLbL5DHKAKlYndzMwVZI=
 github.com/fsnotify/fsnotify v1.5.4/go.mod h1:OVB6XrOHzAwXMpEM7uPOzcehqUV2UqJxmVXmkdnm1bU=
+github.com/fsnotify/fsnotify v1.8.0 h1:dAwr6QBTBZIkG8roQaJjGof0pp0EeF+tNV7YBP3F/8M=
+github.com/fsnotify/fsnotify v1.8.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0=
 github.com/fsouza/fake-gcs-server v0.0.0-20180612165233-e85be23bdaa8/go.mod h1:1/HufuJ+eaDf4KTnYdS6HJMGvMRU8d4cYTuu/1QaBbI=
 github.com/fsouza/fake-gcs-server v1.19.4/go.mod h1:I0/88nHCASqJJ5M7zVF0zKODkYTcuXFW5J5yajsNJnE=
 github.com/fvbommel/sortorder v1.0.1/go.mod h1:uk88iVf1ovNn1iLfgUVU2F9o5eO30ui720w+kxuqRs0=

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

@@ -7,6 +7,7 @@ import (
 	"strings"
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
+	fsnotify "github.com/fsnotify/fsnotify"
 
 	"k8s.io/klog/v2"
 
@@ -25,7 +26,7 @@ var raidedLocalSsdPath = raidedLssdPrefix + raidedLocalSsdName
 
 func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId string) (string, error) {
 	volumeId := req.GetVolumeId()
-	volumeGroupName := getVolumeGroupName(nodeId)
+	volumeGroupName := GetVolumeGroupName(nodeId)
 	mainDevicePath := "/dev/" + volumeGroupName + "/" + getLvName(mainLvSuffix, volumeId)
 	mainLvName := getLvName(mainLvSuffix, volumeId)
 	klog.V(2).Infof("Volume group available on node %v ", volumeGroupName)
@@ -37,17 +38,6 @@ func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId str
 	infoString := strings.TrimSpace(string(info))
 	raidedLocalSsdPath = raidedLssdPrefix + infoString
 
-	vgExists := checkVgExists(volumeGroupName)
-	if vgExists {
-		// Clean up Volume Group before adding the PD
-		reduceVolumeGroup(volumeGroupName, true)
-	} else {
-		err := createVg(volumeGroupName, devicePath, raidedLocalSsdPath)
-		if err != nil {
-			return mainDevicePath, err
-		}
-	}
-
 	// Check if the Physical Volume(PV) is part of some other volume group
 	args := []string{
 		"--select",
@@ -75,7 +65,7 @@ func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId str
 			klog.Errorf("Errored while deactivating VG %v: err: %v: %s", vgNameForPv, err, info)
 		}
 		// CLean up volume group to remove any dangling PV refrences
-		reduceVolumeGroup(vgNameForPv, false)
+		ReduceVolumeGroup(vgNameForPv, false)
 		_, isCached := isCachingSetup(mainLvName)
 		// We will continue to uncache even if it errors to check caching as it is not a terminal issue.
 		if isCached {
@@ -91,7 +81,7 @@ func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId str
 				return "", fmt.Errorf("errored while uncaching main LV. %v: %s", err, info)
 			}
 			// CLean up volume group to remove any dangling PV refrences
-			reduceVolumeGroup(vgNameForPv, false)
+			ReduceVolumeGroup(vgNameForPv, false)
 		}
 		info, err = common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "vgmerge", []string{volumeGroupName, vgNameForPv}...)
 		if err != nil {
@@ -189,7 +179,7 @@ func setupCaching(devicePath string, req *csi.NodeStageVolumeRequest, nodeId str
 	return mainDevicePath, nil
 }
 
-func checkVgExists(volumeGroupName string) bool {
+func CheckVgExists(volumeGroupName string) bool {
 	args := []string{}
 	info, err := common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "vgscan", args...)
 	if err != nil {
@@ -202,8 +192,8 @@ func checkVgExists(volumeGroupName string) bool {
 
 func cleanupCache(volumeId string, nodeId string) error {
 
-	volumeGroupName := getVolumeGroupName(nodeId)
-	if !checkVgExists(volumeGroupName) {
+	volumeGroupName := GetVolumeGroupName(nodeId)
+	if !CheckVgExists(volumeGroupName) {
 		// If volume group doesn't exist then there's nothing to uncache
 		return nil
 	}
@@ -228,7 +218,7 @@ func cleanupCache(volumeId string, nodeId string) error {
 	return nil
 }
 
-func getVolumeGroupName(nodePath string) string {
+func GetVolumeGroupName(nodePath string) string {
 	nodeSlice := strings.Split(nodePath, "/")
 	nodeId := nodeSlice[len(nodeSlice)-1]
 	nodeHash := common.ShortString(nodeId)
@@ -241,7 +231,7 @@ func getLvName(suffix string, volumeId string) string {
 	return fmt.Sprintf("%s-%s", suffix, pvcName)
 }
 
-func createVg(volumeGroupName string, devicePath string, raidedLocalSsds string) error {
+func CreateVg(volumeGroupName string, raidedLocalSsds string) error {
 	args := []string{
 		"--zero",
 		"y",
@@ -263,7 +253,7 @@ func createVg(volumeGroupName string, devicePath string, raidedLocalSsds string)
 	return nil
 }
 
-func reduceVolumeGroup(volumeGroupName string, force bool) {
+func ReduceVolumeGroup(volumeGroupName string, force bool) {
 	args := []string{
 		"--removemissing",
 		volumeGroupName,
@@ -366,3 +356,65 @@ func isCachingSetup(mainLvName string) (error, bool) {
 	}
 	return nil, false
 }
+
+func StartWatcher(nodeName string) {
+	dirToWatch := "/dev/"
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		klog.V(2).ErrorS(err, "errored while creating watcher")
+	}
+	defer watcher.Close()
+
+	// out of the box fsnotify can watch a single file, or a single directory
+	if err := watcher.Add(dirToWatch); err != nil {
+		klog.V(2).ErrorS(err, "errored while adding watcher directory")
+	}
+	errorCh := make(chan error, 1)
+	// Handle the error received from the watcher goroutine
+	go watchDiskDetaches(watcher, nodeName, errorCh)
+
+	select {
+	case err := <-errorCh:
+		klog.Errorf("watcher encountered an error: %v", err)
+	}
+}
+
+func watchDiskDetaches(watcher *fsnotify.Watcher, nodeName string, errorCh chan error) error {
+	for {
+		select {
+		// watch for errors
+		case err := <-watcher.Errors:
+			errorCh <- fmt.Errorf("disk update event errored: %v", err)
+		// watch for events
+		case event := <-watcher.Events:
+			// In case of an event i.e. creation or deletion of any new PV, we update the VG metadata.
+			// This might include some non-LVM changes, no harm in updating metadata multiple times.
+			ReduceVolumeGroup(GetVolumeGroupName(nodeName), true)
+			klog.V(2).Infof("disk attach/detach event %#v\n", event)
+		}
+	}
+}
+
+func ValidateRaidedLSSDinVG(vgName string) error {
+	args := []string{
+		"--noheadings",
+		"-o",
+		"pv_name",
+		"--select",
+		"vg_name=" + vgName,
+	}
+	info, err := common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "pvs", args...)
+	if err != nil {
+		return fmt.Errorf("errored while checking physical volume details %v: %s", err, info)
+		// On error info contains the error message which we cannot use for further steps
+	}
+
+	klog.V(2).Infof("Got PVs %v in VG %v", strings.TrimSpace(string(info)), vgName)
+	if !strings.Contains(string(info), "/dev/md127") {
+		info, err := common.RunCommand("" /* pipedCmd */, "" /* pipedCmdArg */, "vgextend", []string{vgName, "/dev/md127"}...)
+		if err != nil {
+			klog.Errorf("errored while extending VGs %v: %s", err, info)
+		}
+	}
+	return nil
+}