Merge pull request #553 from hungnguyen243/passMachineTypeToGCSFuse

Pass --machine-type and --disable-autoconfig flags from driver to gcsfuse, which are new mountOption args supported by gcsfuse v2.12. This is the first step to enable intelligent defaults for high-performance machine types in gcsfuse v3.

Author: hungnguyen243

cmd/csi_driver/main.go

@@ -48,7 +48,7 @@ var (
 	fuseSocketDir             = flag.String("fuse-socket-dir", "/sockets", "FUSE socket directory")
 	metricsEndpoint           = flag.String("metrics-endpoint", "", "The TCP network address where the Prometheus metrics endpoint will listen (example: `:8080`). The default is empty string, which means that the metrics endpoint is disabled.")
 	maximumNumberOfCollectors = flag.Int("max-metric-collectors", -1, "Maximum number of prometheus metric collectors exporting metrics at a time, less than 0 (e.g -1) means no limit.")
-
+	disableAutoconfig         = flag.Bool("disable-autoconfig", false, "Disable gcsfuse's defaulting based on machine type")
 	// These are set at compile time.
 	version = "unknown"
 )
@@ -126,6 +126,7 @@ func main() {
 		Mounter:               mounter,
 		K8sClients:            clientset,
 		MetricsManager:        mm,
+		DisableAutoconfig:     *disableAutoconfig,
 	}
 
 	gcfsDriver, err := driver.NewGCSDriver(config)

cmd/sidecar_mounter/main.go

@@ -28,6 +28,7 @@ import (
 	"syscall"
 	"time"
 
+	driver "github.com/googlecloudplatform/gcs-fuse-csi-driver/pkg/csi_driver"
 	sidecarmounter "github.com/googlecloudplatform/gcs-fuse-csi-driver/pkg/sidecar_mounter"
 	"github.com/googlecloudplatform/gcs-fuse-csi-driver/pkg/webhook"
 	"k8s.io/klog/v2"
@@ -55,12 +56,24 @@ func main() {
 	mounter := sidecarmounter.New(*gcsfusePath)
 	ctx, cancel := context.WithCancel(context.Background())
 
+	flagsFromDriver := map[string]string{}
+	defaultingFlagFilePath := *volumeBasePath + "/" + driver.FlagFileForDefaultingPath
+	klog.Infof("Checking if defaulting-flag file %q exists", defaultingFlagFilePath)
+	if _, err := os.Stat(defaultingFlagFilePath); err == nil {
+		machineTypeBytes, err := os.ReadFile(defaultingFlagFilePath)
+		if err != nil {
+			klog.Fatalf("failed to read defaulting-flag file: %v", err)
+		}
+		fileContent := string(machineTypeBytes)
+		flagsFromDriver = driver.ParseFlagMapFromFlagFile(fileContent)
+	}
+
 	for _, sp := range socketPaths {
 		// sleep 1.5 seconds before launch the next gcsfuse to avoid
 		// 1. different gcsfuse logs mixed together.
 		// 2. memory usage peak.
 		time.Sleep(1500 * time.Millisecond)
-		mc := sidecarmounter.NewMountConfig(sp)
+		mc := sidecarmounter.NewMountConfig(sp, flagsFromDriver)
 		if mc != nil {
 			if err := mounter.Mount(ctx, mc); err != nil {
 				mc.ErrWriter.WriteMsg(fmt.Sprintf("failed to mount bucket %q for volume %q: %v\n", mc.BucketName, mc.VolumeName, err))

pkg/cloud_provider/clientset/clientset.go

@@ -59,6 +59,7 @@ type Clientset struct {
 }
 
 const GkeMetaDataServerKey = "iam.gke.io/gke-metadata-server-enabled"
+const MachineTypeKey = "node.kubernetes.io/instance-type"
 
 func (c *Clientset) ConfigureNodeLister(nodeName string) {
 	trim := func(obj interface{}) (interface{}, error) {
@@ -81,6 +82,10 @@ func (c *Clientset) ConfigureNodeLister(nodeName string) {
 		if ok {
 			newLabels[GkeMetaDataServerKey] = isGkeMetaDataServerEnabled
 		}
+		machineType, ok := nodeObj.ObjectMeta.Labels[MachineTypeKey]
+		if ok {
+			newLabels[MachineTypeKey] = machineType
+		}
 
 		nodeObj.Spec = corev1.NodeSpec{}
 		nodeObj.Status = corev1.NodeStatus{}

pkg/cloud_provider/clientset/fake.go

@@ -19,13 +19,19 @@ package clientset
 
 import (
 	"context"
+	"strconv"
 
 	"github.com/googlecloudplatform/gcs-fuse-csi-driver/pkg/webhook"
 	authenticationv1 "k8s.io/api/authentication/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+type FakeNodeConfig struct {
+	IsWorkloadIdentityEnabled bool
+	MachineType               string
+}
+
 type FakeClientset struct {
 	fakePod  *corev1.Pod
 	fakeNode *corev1.Node
@@ -35,7 +41,7 @@ func NewFakeClientset() *FakeClientset {
 	fakeClientSet := &FakeClientset{}
 	// Default setting for most unit tests is pod doesn't use host network & workload identity is enabled on the node
 	fakeClientSet.CreatePod( /*hostNetworkEnabled */ false)
-	fakeClientSet.CreateNode( /* isWorkloadIdentityEnabledOnNode */ true)
+	fakeClientSet.CreateNode(FakeNodeConfig{IsWorkloadIdentityEnabled: true})
 
 	return fakeClientSet
 }
@@ -74,16 +80,18 @@ func (c *FakeClientset) CreatePod(hostNetworkEnabled bool) {
 	}
 }
 
-func (c *FakeClientset) CreateNode(isWorkloadIdentityEnabled bool) {
+func (c *FakeClientset) CreateNode(nodeConfig FakeNodeConfig) {
 	c.fakeNode = &corev1.Node{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:   "",
 			Labels: map[string]string{},
 		},
 	}
 
-	if isWorkloadIdentityEnabled {
-		c.fakeNode.Labels[GkeMetaDataServerKey] = "true"
+	c.fakeNode.Labels[GkeMetaDataServerKey] = strconv.FormatBool(nodeConfig.IsWorkloadIdentityEnabled)
+	c.fakeNode.Labels[MachineTypeKey] = nodeConfig.MachineType
+	if c.fakeNode.Labels[MachineTypeKey] == "" {
+		c.fakeNode.Labels[MachineTypeKey] = "e2-medium"
 	}
 }
 

pkg/csi_driver/gcs_fuse_driver.go

@@ -45,6 +45,7 @@ type GCSDriverConfig struct {
 	Mounter               mount.Interface
 	K8sClients            clientset.Interface
 	MetricsManager        metrics.Manager
+	DisableAutoconfig     bool
 }
 
 type GCSDriver struct {

pkg/csi_driver/node.go

@@ -20,6 +20,7 @@ package driver
 import (
 	"fmt"
 	"os"
+	"strconv"
 	"strings"
 	"time"
 
@@ -182,6 +183,20 @@ func (s *nodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublish
 		}
 	}
 
+	// Only pass mountOptions flags for defaulting if sidecar container is managed and satisifies min version requirement
+	if s.shouldPassDefaultingFlags(pod) {
+		shouldDisableAutoConfig := s.driver.config.DisableAutoconfig
+		machineType, ok := node.Labels[clientset.MachineTypeKey]
+		if ok {
+			flagMap := map[string]string{"machine-type": machineType, "disable-autoconfig": strconv.FormatBool(shouldDisableAutoConfig)}
+			if err := PutFlagsFromDriverToTargetPath(flagMap, targetPath, FlagFileForDefaultingPath); err != nil {
+				return nil, status.Error(codes.Internal, err.Error())
+			}
+		} else {
+			klog.Warningf("Unable to fetch target node %v's machine type", node.Name)
+		}
+	}
+
 	// Check if there is any error from the gcsfuse
 	code, err := checkGcsFuseErr(isInitContainer, pod, targetPath)
 	if code != codes.OK {
@@ -330,3 +345,16 @@ func (s *nodeServer) shouldStartTokenServer(pod *corev1.Pod) bool {
 
 	return tokenVolumeInjected && sidecarVersionSupported
 }
+
+func (s *nodeServer) shouldPassDefaultingFlags(pod *corev1.Pod) bool {
+	var sidecarVersionSupported bool
+	for _, container := range pod.Spec.InitContainers {
+		if container.Name == webhook.GcsFuseSidecarName {
+			sidecarVersionSupported = isSidecarVersionSupportedForGivenFeature(container.Image, AutoconfigDefaultingSidecarMinVersion)
+
+			break
+		}
+	}
+
+	return sidecarVersionSupported
+}

pkg/csi_driver/node_test.go

@@ -247,7 +247,7 @@ func TestNodePublishVolumeWIDisabledOnNode(t *testing.T) {
 	}
 	for _, test := range cases {
 		fakeClientSet := &clientset.FakeClientset{}
-		fakeClientSet.CreateNode( /* workloadIdentityEnabled */ test.workloadIdentityEnabledOnNode)
+		fakeClientSet.CreateNode( /* workloadIdentityEnabled */ clientset.FakeNodeConfig{IsWorkloadIdentityEnabled: test.workloadIdentityEnabledOnNode})
 		fakeClientSet.CreatePod( /* hostNetworkEnabled */ test.hostNetworkEnabledOnPod)
 		testEnv := initTestNodeServerWithCustomClientset(t, fakeClientSet)
 

pkg/csi_driver/utils.go

@@ -66,6 +66,9 @@ const (
 	VolumeContextKeyEphemeral           = "csi.storage.k8s.io/ephemeral"
 	VolumeContextKeyBucketName          = "bucketName"
 	tokenServerSidecarMinVersion        = "v1.12.2-gke.0" // #nosec G101
+	// TODO: Update with actual minimum sidecar version after the first feature release
+	AutoconfigDefaultingSidecarMinVersion = "v1.99.0-gke.0"
+	FlagFileForDefaultingPath             = "flags-for-defaulting"
 )
 
 var volumeIDRegEx = regexp.MustCompile(`:.*$`)
@@ -473,22 +476,73 @@ func getSidecarContainerStatus(isInitContainer bool, pod *corev1.Pod) (*corev1.C
 }
 
 func isSidecarVersionSupportedForTokenServer(imageName string) bool {
+	return isSidecarVersionSupportedForGivenFeature(imageName, tokenServerSidecarMinVersion)
+}
+
+func isSidecarVersionSupportedForGivenFeature(imageName string, sidecarMinSupportedVersion string) bool {
 	managedSidecarPattern := `.*/gke-release(-staging)?/gcs-fuse-csi-driver-sidecar-mounter:v\d+.\d+.\d+-gke\.\d+.*`
 	re := regexp.MustCompile(managedSidecarPattern)
 	isManagedSidecar := re.MatchString(imageName)
 
 	if !isManagedSidecar {
 		klog.Infof("mountOptions should not be passed because this is a private sidecar image %q", imageName)
-
 		return false
 	}
 	imageVersion := strings.Split(strings.Split(imageName, ":")[1], "@")[0]
 	klog.Infof("sidecar image version: %v", imageVersion)
-	if semver.Compare(imageVersion, tokenServerSidecarMinVersion) >= 0 {
+	if semver.Compare(imageVersion, sidecarMinSupportedVersion) >= 0 {
 		klog.Infof("sidecar version is supported for token server")
-
 		return true
 	}
 
 	return false
 }
+
+func PutFlagsFromDriverToTargetPath(flagMap map[string]string, targetPath string, fileName string) error {
+	emptyDirBasePath, err := util.PrepareEmptyDir(targetPath, true)
+	if err != nil {
+		return fmt.Errorf("failed to get emptyDir path: %w", err)
+	}
+
+	absolutePath := filepath.Dir(emptyDirBasePath) + "/" + fileName
+	klog.V(4).Infof("Writing flags needed for gcsfuse defaulting logic to file %q: %v", absolutePath, flagMap)
+
+	f, err := os.Create(absolutePath)
+	if err != nil {
+		return fmt.Errorf("failed to create defaulting-flag file: %w", err)
+	}
+	content := prepareFileContentFromFlagMap(flagMap)
+	if _, err := f.WriteString(content); err != nil {
+		return fmt.Errorf("failed to write defaulting-flag file: %w", err)
+	}
+
+	f.Close()
+
+	return nil
+}
+
+func prepareFileContentFromFlagMap(flagMap map[string]string) string {
+	var sb strings.Builder
+	for key, value := range flagMap {
+		sb.WriteString(key)
+		sb.WriteString(":")
+		sb.WriteString(value)
+		sb.WriteString("\n")
+	}
+	return sb.String()
+}
+
+func ParseFlagMapFromFlagFile(flagFileContent string) map[string]string {
+	configFlags := make(map[string]string)
+	lines := strings.Split(flagFileContent, "\n")
+	for _, line := range lines {
+		if line == "" { // Skip empty lines
+			continue
+		}
+		parts := strings.Split(line, ":")
+		key := parts[0]
+		value := parts[1]
+		configFlags[key] = value
+	}
+	return configFlags
+}

pkg/csi_driver/utils_test.go

@@ -156,6 +156,47 @@ func TestIsSidecarVersionSupportedForTokenServer(t *testing.T) {
 	})
 }
 
+func TestIsSidecarVersionSupportedForDefaultingFlags(t *testing.T) {
+	t.Parallel()
+	t.Run("checking if sidecar version is supported for token server", func(t *testing.T) {
+		t.Parallel()
+		testCases := []struct {
+			name              string
+			imageName         string
+			expectedSupported bool
+		}{
+			{
+				name:              "should return true for supported sidecar version",
+				imageName:         "us-central1-artifactregistry.gcr.io/gke-release/gke-release/gcs-fuse-csi-driver-sidecar-mounter:v1.99.0-gke.2@sha256:abcd",
+				expectedSupported: true,
+			},
+			{
+				name:              "should return true for supported sidecar version in staging gcr",
+				imageName:         "gcr.io/gke-release-staging/gcs-fuse-csi-driver-sidecar-mounter:v1.99.0-gke.0@sha256:abcd",
+				expectedSupported: true,
+			},
+			{
+				name:              "should return false for unsupported sidecar version",
+				imageName:         "us-central1-artifactregistry.gcr.io/gke-release/gke-release/gcs-fuse-csi-driver-sidecar-mounter:v1.14.0-gke.1@sha256:abcd",
+				expectedSupported: false,
+			},
+			{
+				name:              "should return false for private sidecar",
+				imageName:         "customer.gcr.io/dir/gcs-fuse-csi-driver-sidecar-mounter:v1.12.2-gke.0@sha256:abcd",
+				expectedSupported: false,
+			},
+		}
+
+		for _, tc := range testCases {
+			t.Logf("test case: %s", tc.name)
+			actual := isSidecarVersionSupportedForGivenFeature(tc.imageName, AutoconfigDefaultingSidecarMinVersion)
+			if actual != tc.expectedSupported {
+				t.Errorf("Got supported %v, but expected %v", actual, tc.expectedSupported)
+			}
+		}
+	})
+}
+
 func TestParseVolumeAttributes(t *testing.T) {
 	t.Parallel()
 	t.Run("parsing volume attributes into mount options", func(t *testing.T) {

pkg/sidecar_mounter/sidecar_mounter_config.go

@@ -88,14 +88,15 @@ var boolFlags = map[string]bool{
 	"debug_http":                    true,
 	"debug_invariants":              true,
 	"debug_mutex":                   true,
+	"disable-autoconfig":            true,
 }
 
 // Fetch the following information from a given socket path:
 // 1. Pod volume name
 // 2. The file descriptor
 // 3. GCS bucket name
 // 4. Mount options passing to gcsfuse (passed by the csi mounter).
-func NewMountConfig(sp string) *MountConfig {
+func NewMountConfig(sp string, flagMapFromDriver map[string]string) *MountConfig {
 	// socket path pattern: /gcsfuse-tmp/.volumes/<volume-name>/socket
 	tempDir := filepath.Dir(sp)
 	volumeName := filepath.Base(tempDir)
@@ -143,6 +144,7 @@ func NewMountConfig(sp string) *MountConfig {
 	}
 
 	mc.prepareMountArgs()
+	mergeFlags(mc.ConfigFileFlagMap, flagMapFromDriver)
 	if err := mc.prepareConfigFile(); err != nil {
 		mc.ErrWriter.WriteMsg(fmt.Sprintf("failed to create config file %q: %v", mc.ConfigFile, err))
 
@@ -152,6 +154,16 @@ func NewMountConfig(sp string) *MountConfig {
 	return &mc
 }
 
+func mergeFlags(mountConfigFlagMap map[string]string, driverFlagMap map[string]string) {
+	for key, value := range driverFlagMap {
+		_, ok := mountConfigFlagMap[key]
+		// Only overwrite values not set in mountConfigMap
+		if !ok {
+			mountConfigFlagMap[key] = value
+		}
+	}
+}
+
 func (mc *MountConfig) prepareMountArgs() {
 	flagMap := map[string]string{
 		"app-name":    GCSFuseAppName,