Half-made changes

Author: julianKatz

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

@@ -27,6 +27,8 @@ import (
 	"strings"
 	"time"
 
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
 	"k8s.io/klog/v2"
 	"k8s.io/utils/strings/slices"
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
@@ -239,19 +241,27 @@ func handle() {
 		if err != nil {
 			klog.Fatalf("Failed to get safe mounter: %v", err.Error())
 		}
+
 		deviceUtils := deviceutils.NewDeviceUtils()
 		statter := mountmanager.NewStatter(mounter)
 		meta, err := metadataservice.NewMetadataService()
 		if err != nil {
 			klog.Fatalf("Failed to set up metadata service: %v", err.Error())
 		}
+
+		kubeClient, err := instantiateKubeClient()
+		if err != nil {
+			klog.Fatalf("Failed to instantiate Kubernetes client: %v", err)
+		}
 		nsArgs := driver.NodeServerArgs{
 			EnableDeviceInUseCheck:   *enableDeviceInUseCheck,
 			DeviceInUseTimeout:       *deviceInUseTimeout,
 			EnableDataCache:          *enableDataCacheFlag,
 			DataCacheEnabledNodePool: isDataCacheEnabledNodePool(ctx, *nodeName),
+			KubeClient:               kubeClient,
 		}
 		nodeServer = driver.NewNodeServer(gceDriver, mounter, deviceUtils, meta, statter, nsArgs)
+
 		if *maxConcurrentFormatAndMount > 0 {
 			nodeServer = nodeServer.WithSerializedFormatAndMount(*formatAndMountTimeout, *maxConcurrentFormatAndMount)
 		}
@@ -288,6 +298,19 @@ func handle() {
 	gceDriver.Run(*endpoint, *grpcLogCharCap, *enableOtelTracing, metricsManager)
 }
 
+func instantiateKubeClient() (*kubernetes.Clientset, error) {
+	klog.V(2).Infof("Setting up kubeClient")
+	cfg, err := rest.InClusterConfig()
+	if err != nil {
+		return nil, fmt.Errorf("failed to create REST Config for k8s client: %w", err)
+	}
+	kubeClient, err := kubernetes.NewForConfig(cfg)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create k8s client: %w", err)
+	}
+	return kubeClient, nil
+}
+
 func notEmpty(v string) bool {
 	return v != ""
 }

pkg/common/constants.go

@@ -17,8 +17,9 @@ limitations under the License.
 package common
 
 const (
+	TopologyKeyPrefix = "topology.gke.io/"
 	// Keys for Topology. This key will be shared amongst drivers from GCP
-	TopologyKeyZone = "topology.gke.io/zone"
+	TopologyKeyZone = TopologyKeyPrefix + "zone"
 
 	// VolumeAttributes for Partition
 	VolumeAttributePartition = "partition"
@@ -32,28 +33,4 @@ const (
 
 	// Label that is set on a disk when it is used by a 'multi-zone' VolumeHandle
 	MultiZoneLabel = "goog-gke-multi-zone"
-
-	// GCE Access Modes that are valid for hyperdisks only.
-	GCEReadOnlyManyAccessMode  = "READ_ONLY_MANY"
-	GCEReadWriteManyAccessMode = "READ_WRITE_MANY"
-	GCEReadWriteOnceAccessMode = "READ_WRITE_SINGLE"
-
-	// 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"
-	// Node name for E2E tests
-	TestNode = "test-node-csi-e2e"
-
-	// Default LSSD count for datacache E2E tests
-	LocalSSDCountForDataCache = 2
-
-	// Node label for Data Cache (only applicable to GKE nodes)
-	NodeLabelPrefix         = "cloud.google.com/%s"
-	DataCacheLssdCountLabel = "gke-data-cache-disk"
 )

pkg/common/utils.go

@@ -20,7 +20,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"hash/fnv"
 	"net/http"
 	"regexp"
 	"slices"
@@ -79,7 +78,8 @@ const (
 	// Full or partial URL of the zone resource, in the format:
 	//   projects/{project}/zones/{zone}
 	zoneURIPattern = "projects/[^/]+/zones/([^/]+)$"
-	alphanums      = "bcdfghjklmnpqrstvwxz2456789"
+
+	gkeTopologyLabelPrefix = "topology.gke.io/"
 )
 
 var (
@@ -103,8 +103,6 @@ var (
 		http.StatusConflict:        codes.FailedPrecondition,
 	}
 
-	validDataCacheMode = []string{DataCacheModeWriteBack, DataCacheModeWriteThrough}
-
 	// Regular expressions for validating parent_id, key and value of a resource tag.
 	regexParent = regexp.MustCompile(`(^[1-9][0-9]{0,31}$)|(^[a-z][a-z0-9-]{4,28}[a-z0-9]$)`)
 	regexKey    = regexp.MustCompile(`^[a-zA-Z0-9]([0-9A-Za-z_.-]{0,61}[a-zA-Z0-9])?$`)
@@ -396,15 +394,6 @@ func ConvertMiStringToInt64(str string) (int64, error) {
 	return volumehelpers.RoundUpToMiB(quantity)
 }
 
-// ConvertGiStringToInt64 converts a GiB string to int64
-func ConvertGiStringToInt64(str string) (int64, error) {
-	quantity, err := resource.ParseQuantity(str)
-	if err != nil {
-		return -1, err
-	}
-	return volumehelpers.RoundUpToGiB(quantity)
-}
-
 // ConvertStringToBool converts a string to a boolean.
 func ConvertStringToBool(str string) (bool, error) {
 	switch strings.ToLower(str) {
@@ -697,29 +686,6 @@ func VolumeIdAsMultiZone(volumeId string) (string, error) {
 	return strings.Join(splitId, "/"), nil
 }
 
-func StringInSlice(s string, list []string) bool {
-	for _, v := range list {
-		if v == s {
-			return true
-		}
-	}
-	return false
-}
-
-func ValidateDataCacheMode(s string) error {
-	if StringInSlice(s, validDataCacheMode) {
-		return nil
-	}
-	return fmt.Errorf("invalid data-cache-mode %s. Only \"writeback\" and \"writethrough\" is a valid input", s)
-}
-
-func ValidateNonNegativeInt(n int64) error {
-	if n <= 0 {
-		return fmt.Errorf("Input should be set to > 0, got %d", n)
-	}
-	return nil
-}
-
 // NewLimiter returns a token bucket based request rate limiter after initializing
 // the passed values for limit, burst (or token bucket) size. If opted for emptyBucket
 // all initial tokens are reserved for the first burst.
@@ -733,19 +699,10 @@ func NewLimiter(limit, burst int, emptyBucket bool) *rate.Limiter {
 	return limiter
 }
 
-func IsHyperdisk(diskType string) bool {
-	return strings.HasPrefix(diskType, "hyperdisk-")
-}
+func IsGKETopologyLabel(key string) bool {
+	// This is the actual code
+	// return strings.HasPrefix(key, gkeTopologyLabelPrefix)
 
-// 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)
+	// More permissive code for testing
+	return strings.HasPrefix(key, "topology.gke")
 }

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

@@ -157,6 +157,7 @@ func NewNodeServer(gceDriver *GCEDriver, mounter *mount.SafeFormatAndMount, devi
 		deviceInUseErrors:        newDeviceErrMap(args.DeviceInUseTimeout),
 		EnableDataCache:          args.EnableDataCache,
 		DataCacheEnabledNodePool: args.DataCacheEnabledNodePool,
+		KubeClient:               args.KubeClient,
 	}
 }
 

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

@@ -30,6 +30,8 @@ import (
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/klog/v2"
 	"k8s.io/mount-utils"
 
@@ -41,13 +43,13 @@ import (
 )
 
 type GCENodeServer struct {
-	Driver                   *GCEDriver
-	Mounter                  *mount.SafeFormatAndMount
-	DeviceUtils              deviceutils.DeviceUtils
-	VolumeStatter            mountmanager.Statter
-	MetadataService          metadataservice.MetadataService
-	EnableDataCache          bool
-	DataCacheEnabledNodePool bool
+	Driver          *GCEDriver
+	Mounter         *mount.SafeFormatAndMount
+	DeviceUtils     deviceutils.DeviceUtils
+	VolumeStatter   mountmanager.Statter
+	MetadataService metadataservice.MetadataService
+
+	kubeClient *kubernetes.Clientset
 
 	// A map storing all volumes with ongoing operations so that additional operations
 	// for that same volume (as defined by VolumeID) return an Aborted error
@@ -80,10 +82,6 @@ type NodeServerArgs struct {
 	EnableDeviceInUseCheck bool
 
 	DeviceInUseTimeout time.Duration
-
-	EnableDataCache bool
-
-	DataCacheEnabledNodePool bool
 }
 
 var _ csi.NodeServer = &GCENodeServer{}
@@ -300,7 +298,6 @@ func (ns *GCENodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStage
 	volumeID := req.GetVolumeId()
 	stagingTargetPath := req.GetStagingTargetPath()
 	volumeCapability := req.GetVolumeCapability()
-	nodeId := ns.MetadataService.GetName()
 	if len(volumeID) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "NodeStageVolume Volume ID must be provided")
 	}
@@ -334,32 +331,12 @@ func (ns *GCENodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStage
 		partition = part
 	}
 	devicePath, err := getDevicePath(ns, volumeID, partition)
+
 	if err != nil {
 		return nil, status.Error(codes.Internal, fmt.Sprintf("Error when getting device path: %v", err.Error()))
 	}
 
-	klog.Infof("Successfully found attached GCE PD %q at device path %s.", volumeKey.Name, devicePath)
-
-	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)
-		if configError != nil {
-			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("The Data Cache PVC is scheduled on an incompatible node pool. Please select a node pool with data cache configured: %v", configError.Error()))
-		}
-		devicePath, err = setupCaching(devFsPath, req, nodeId)
-		if err != nil {
-			return nil, status.Error(codes.DataLoss, fmt.Sprintf("Error setting up cache: %v", err.Error()))
-		}
-	}
+	klog.V(4).Infof("Successfully found attached GCE PD %q at device path %s.", volumeKey.Name, devicePath)
 
 	// Part 2: Check if mount already exists at stagingTargetPath
 	if ns.isVolumePathMounted(stagingTargetPath) {
@@ -513,15 +490,6 @@ func (ns *GCENodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUns
 		ns.deviceInUseErrors.deleteDevice(volumeID)
 	}
 
-	// The NodeUnstageVolume does not have any volume or publish context, we need to get the info from LVM locally
-	// Check if cache group cache-{volumeID} exist in LVM
-	if ns.EnableDataCache {
-		nodeId := ns.MetadataService.GetName()
-		err := cleanupCache(volumeID, nodeId)
-		if err != nil {
-			return nil, status.Errorf(codes.DataLoss, "Failed to cleanup cache for volume %s: %v", volumeID, err)
-		}
-	}
 	klog.V(4).Infof("NodeUnstageVolume succeeded on %v from %s", volumeID, stagingTargetPath)
 	return &csi.NodeUnstageVolumeResponse{}, nil
 }
@@ -556,22 +524,60 @@ func (ns *GCENodeServer) NodeGetCapabilities(ctx context.Context, req *csi.NodeG
 }
 
 func (ns *GCENodeServer) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoRequest) (*csi.NodeGetInfoResponse, error) {
+	labels, err := ns.gkeTopologyLabels(ctx, ns.MetadataService.GetName())
+	if err != nil {
+		// Perhaps we don't want to fail here.  We are introducing a new
+		// dependency and we might be better off allowing this failure to
+		// happen and moving on to retrieve the zone from GCE MDS.
+		return nil, err
+	}
+
+	labels[common.TopologyKeyZone] = ns.MetadataService.GetZone()
+
+	// Each "Topology" struct will later be translated into an individual
+	// 'matchExpressions' block in the PV's NodeAffinity.  Because we always
+	// need to match on both the zone AND the disk type, both the zone and the
+	// supported disks belong as segments on a single Topology.
 	top := &csi.Topology{
-		Segments: map[string]string{common.TopologyKeyZone: ns.MetadataService.GetZone()},
+		Segments: labels,
 	}
 
 	nodeID := common.CreateNodeID(ns.MetadataService.GetProject(), ns.MetadataService.GetZone(), ns.MetadataService.GetName())
-
 	volumeLimits, err := ns.GetVolumeLimits()
 
 	resp := &csi.NodeGetInfoResponse{
 		NodeId:             nodeID,
 		MaxVolumesPerNode:  volumeLimits,
 		AccessibleTopology: top,
 	}
+
+	klog.V(2).Infof("Returning NodeGetInfoResponse: %+v", resp)
+
 	return resp, err
 }
 
+// gkeTopologyLabels retrieves the node labels with the prefix
+// `topology.gke.io/`.
+func (ns *GCENodeServer) gkeTopologyLabels(ctx context.Context, nodeName string) (map[string]string, error) {
+	klog.V(2).Infof("Retrieving node topology labels for node %q", nodeName)
+
+	node, err := ns.kubeClient.CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})
+	if err != nil {
+		// We should retry instead.  Need to figure out how much wrong-ness can be tolerated and how often CSINode gets refreshed.
+		return nil, err
+	}
+
+	topology := make(map[string]string)
+	for k, v := range node.GetLabels() {
+		if common.IsGKETopologyLabel(k) {
+			klog.V(2).Infof("Including node topology label %q=%q", k, v)
+			topology[k] = v
+		}
+	}
+
+	return topology, nil
+}
+
 func (ns *GCENodeServer) NodeGetVolumeStats(ctx context.Context, req *csi.NodeGetVolumeStatsRequest) (*csi.NodeGetVolumeStatsResponse, error) {
 	if len(req.VolumeId) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "NodeGetVolumeStats volume ID was empty")