Add start gcsfuse

Author: chrisThePattyEater

cmd/csi_driver/main.go

@@ -24,6 +24,7 @@ import (
 	"net/http/pprof"
 	"os"
 	"os/signal"
+	"path/filepath"
 	"syscall"
 	"time"
 
@@ -246,6 +247,10 @@ func main() {
 		},
 		SharedMountOptions: &driver.SharedMountOptions{
 			MounterPodImage: *mounterPodImage,
+			FuseSocketDir:   *fuseSocketDir,
+			EmptyDirBasePath: func(podUID string) string {
+				return filepath.Join(util.KubeletDir, "pods", podUID, "volumes", "kubernetes.io~empty-dir", util.SidecarContainerTmpVolumeName)
+			},
 		},
 	}
 

pkg/csi_driver/gcs_fuse_driver.go

@@ -21,14 +21,12 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"path/filepath"
 
 	"github.com/googlecloudplatform/gcs-fuse-csi-driver/pkg/cloud_provider/auth"
 	"github.com/googlecloudplatform/gcs-fuse-csi-driver/pkg/cloud_provider/clientset"
 	"github.com/googlecloudplatform/gcs-fuse-csi-driver/pkg/cloud_provider/storage"
 	"github.com/googlecloudplatform/gcs-fuse-csi-driver/pkg/metrics"
 	"github.com/googlecloudplatform/gcs-fuse-csi-driver/pkg/profiles"
-	"github.com/googlecloudplatform/gcs-fuse-csi-driver/pkg/util"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -54,6 +52,9 @@ type GoMemLimitOptions struct {
 
 type SharedMountOptions struct {
 	MounterPodImage string
+	FuseSocketDir   string
+	// Needed to override the mounter pods emptydir base path, otherwise tests will try to write to var/lib/kubelet which it won't have access to.
+	EmptyDirBasePath func(podUID string) string
 }
 
 type GCSDriverFeatureOptions struct {
@@ -82,7 +83,6 @@ type GCSDriverConfig struct {
 	FeatureOptions                 *GCSDriverFeatureOptions
 	AssumeGoodSidecarVersion       bool
 	UniverseDomain                 string
-	EmptyDirBasePath               func(podUID string) string
 }
 
 type GCSDriver struct {
@@ -112,11 +112,6 @@ func NewGCSDriver(config *GCSDriverConfig, recorder record.EventRecorder) (*GCSD
 	if !config.RunController && !config.RunNode {
 		return nil, errors.New("must run at least one controller or node service")
 	}
-	if config.EmptyDirBasePath == nil {
-		config.EmptyDirBasePath = func(podUID string) string {
-			return filepath.Join(util.KubeletDir, "pods", podUID, "volumes", "kubernetes.io~empty-dir", util.SidecarContainerTmpVolumeName)
-		}
-	}
 
 	driver := &GCSDriver{
 		config:   config,

pkg/csi_driver/gcs_fuse_driver_test.go

@@ -47,6 +47,7 @@ func initTestDriver(t *testing.T, fm *mount.FakeMounter, clientset clientset.Int
 			FeatureGCSFuseProfiles: &FeatureGCSFuseProfiles{},
 			SharedMountOptions: &SharedMountOptions{
 				MounterPodImage: testImage,
+				FuseSocketDir:   "/tmp/fuse-sockets",
 			},
 		},
 	}
@@ -65,19 +66,25 @@ func initTestDriver(t *testing.T, fm *mount.FakeMounter, clientset clientset.Int
 func initTestDriverWithCustomNodeServer(t *testing.T, fm *mount.FakeMounter, clientSet *clientset.FakeClientset, wiNodeLabelCheck bool) *GCSDriver {
 	t.Helper()
 	config := &GCSDriverConfig{
-		Name:                     "test-driver",
-		NodeID:                   "test-node",
-		Version:                  "test-version",
-		RunController:            true,
-		RunNode:                  true,
-		StorageServiceManager:    storage.NewFakeServiceManager(),
-		TokenManager:             auth.NewFakeTokenManager(),
-		Mounter:                  fm,
-		NetworkManager:           &fakeNetworkManager{},
-		K8sClients:               clientSet,
-		MetricsManager:           metrics.NewFakeMetricsManager(),
-		WINodeLabelCheck:         wiNodeLabelCheck,
-		FeatureOptions:           &GCSDriverFeatureOptions{FeatureGCSFuseProfiles: &FeatureGCSFuseProfiles{}},
+		Name:                  "test-driver",
+		NodeID:                "test-node",
+		Version:               "test-version",
+		RunController:         true,
+		RunNode:               true,
+		StorageServiceManager: storage.NewFakeServiceManager(),
+		TokenManager:          auth.NewFakeTokenManager(),
+		Mounter:               fm,
+		NetworkManager:        &fakeNetworkManager{},
+		K8sClients:            clientSet,
+		MetricsManager:        metrics.NewFakeMetricsManager(),
+		WINodeLabelCheck:      wiNodeLabelCheck,
+		FeatureOptions: &GCSDriverFeatureOptions{
+			FeatureGCSFuseProfiles: &FeatureGCSFuseProfiles{},
+			SharedMountOptions: &SharedMountOptions{
+				MounterPodImage: testImage,
+				FuseSocketDir:   "/tmp/fuse-sockets",
+			},
+		},
 		AssumeGoodSidecarVersion: true,
 	}
 	driver, err := NewGCSDriver(config, nil)

pkg/csi_driver/node.go

@@ -20,6 +20,7 @@ package driver
 import (
 	"fmt"
 	"os"
+	"path/filepath"
 	"strconv"
 	"strings"
 	"time"
@@ -32,7 +33,9 @@ import (
 	"github.com/googlecloudplatform/gcs-fuse-csi-driver/pkg/webhook"
 	"golang.org/x/net/context"
 	"golang.org/x/time/rate"
+	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/status"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -727,13 +730,75 @@ func (s *nodeServer) executeNodeStageVolume(ctx context.Context, req *csi.NodeSt
 	}
 
 	// Wait for the mounter pod grpc server to be ready.
-	if err := waitForMounterServer(ctx, clientset, podNamespace, podName, string(pod.UID), s.driver.config.EmptyDirBasePath); err != nil {
+	if err := waitForMounterServer(ctx, clientset, podNamespace, podName, string(pod.UID), s.driver.config.FeatureOptions.SharedMountOptions.EmptyDirBasePath); err != nil {
+		return nil, err
+	}
+
+	podUID := string(pod.UID)
+
+	// Send GRPC to mounter pod to start GCSFuse.
+	if err := s.mountToNode(ctx, podUID, stagingPath, req.GetVolumeId()); err != nil {
 		return nil, err
 	}
 
-	// TODO(FUECHR) Add start gcsfuse flow.
 	klog.Infof("Mounter pod %s/%s is running and staging path %s is mounted", podNamespace, podName, stagingPath)
 
 	klog.Infof("NodeStageVolume succeeded on staging path %q for volume %q", stagingPath, req.GetVolumeId())
 	return &csi.NodeStageVolumeResponse{}, nil
 }
+
+// mountToNode connects to the mounter server, at which point it initializes the GCSFuse process.
+func (s *nodeServer) mountToNode(ctx context.Context, podUID, stagingPath, volumeID string) error {
+	if s.driver.config.FeatureOptions == nil || s.driver.config.FeatureOptions.SharedMountOptions == nil {
+		return status.Errorf(codes.Internal, "shared mount options are not fully configured")
+	}
+
+	if s.driver.config.FeatureOptions.SharedMountOptions.EmptyDirBasePath == nil {
+		return status.Errorf(codes.Internal, "empty dir base path must be provided for shared mount")
+	}
+	emptyDirBasePath := s.driver.config.FeatureOptions.SharedMountOptions.EmptyDirBasePath(podUID)
+	socketFile := filepath.Join(emptyDirBasePath, mounterPodSocketFile)
+
+	// Create a symlink to bypass the 108-character limit for Unix domain sockets
+	// when dialing the connection from the Node Server.
+	if s.driver.config.FeatureOptions.SharedMountOptions.FuseSocketDir == "" {
+		return status.Errorf(codes.Internal, "fuse socket dir must be provided for shared mount")
+	}
+	symlink := filepath.Join(s.driver.config.FeatureOptions.SharedMountOptions.FuseSocketDir, mounterPodSocketDir, podUID)
+	if err := os.MkdirAll(filepath.Dir(symlink), 0750); err != nil {
+		return status.Errorf(codes.Internal, "failed to create dir for symlink %q: %v", symlink, err)
+	}
+
+	if err := os.Remove(symlink); err != nil && !os.IsNotExist(err) {
+		klog.Errorf("failed to remove stale symlink %q: %v", symlink, err)
+	}
+
+	if err := os.Symlink(socketFile, symlink); err != nil {
+		return status.Errorf(codes.Internal, "failed to create symlink to %q: %v", socketFile, err)
+	}
+	defer os.Remove(symlink)
+
+	// Connect to the socket using the short symlink path.
+	socketPath := fmt.Sprintf("unix:%s", symlink)
+	conn, err := grpc.NewClient(socketPath, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		klog.Errorf("Failed to connect to the server: %v", err)
+		return status.Errorf(codes.Internal, "failed to connect to the mounter pod grpc server: %v", err)
+	}
+	klog.Infof("Connected to MounterServer at %s", socketPath)
+	defer conn.Close()
+
+	/*
+		TODO(FUECHR): Implement the mounter client and mount request once we have a mounter service defined.
+		c := mounter.NewMounterClient(conn)
+		if _, err := c.Mount(ctx, &mounter.MountRequest{
+			Mountpoint: stagingPath,
+			VolumeID:   volumeID,
+		}); err != nil {
+			return status.Errorf(codes.Internal, "failed to mount: %v", err)
+		}
+	*/
+
+	klog.Infof("Mount succeeded at staging target path %s", stagingPath)
+	return nil
+}

pkg/csi_driver/node_test.go

@@ -325,7 +325,7 @@ func TestExecuteNodeStageVolume(t *testing.T) {
 			if !ok {
 				t.Fatalf("Failed to cast NodeServer to *nodeServer")
 			}
-			ns.driver.config.EmptyDirBasePath = func(uid string) string {
+			ns.driver.config.FeatureOptions.SharedMountOptions.EmptyDirBasePath = func(uid string) string {
 				return filepath.Join(kubeletDir, "pods", uid, "volumes", "kubernetes.io~empty-dir", util.SidecarContainerTmpVolumeName)
 			}
 
@@ -982,7 +982,7 @@ func TestNodeStageVolume(t *testing.T) {
 			if !ok {
 				t.Fatalf("Failed to cast NodeServer to *nodeServer")
 			}
-			ns.driver.config.EmptyDirBasePath = func(uid string) string {
+			ns.driver.config.FeatureOptions.SharedMountOptions.EmptyDirBasePath = func(uid string) string {
 				return filepath.Join(kubeletDir, "pods", uid, "volumes", "kubernetes.io~empty-dir", util.SidecarContainerTmpVolumeName)
 			}
 
@@ -1001,6 +1001,58 @@ func TestNodeStageVolume(t *testing.T) {
 	}
 }
 
+func TestMountToNode(t *testing.T) {
+	tests := []struct {
+		name    string
+		wantErr bool
+	}{
+		{
+			name:    "successful mount to node - should succeed",
+			wantErr: false,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			podUID := "test-pod-uid"
+			volumeID := "test-volume-id"
+			stagingPath := filepath.Join(t.TempDir(), "var/lib/kubelet/pods", podUID, "volumes/kubernetes.io~csi", volumeID, "mount")
+
+			testEnv := initTestNodeServer(t)
+			ns, ok := testEnv.ns.(*nodeServer)
+			if !ok {
+				t.Fatalf("Failed to cast NodeServer to *nodeServer")
+			}
+
+			emptyDirBasePath, err := util.PrepareEmptyDir(stagingPath, true)
+			if err != nil {
+				t.Fatalf("failed to prepare emptyDir path: %v", err)
+			}
+			ns.driver.config.FeatureOptions.SharedMountOptions.EmptyDirBasePath = func(uid string) string {
+				return emptyDirBasePath
+			}
+
+			// Create a dummy socket file for the client to connect to
+			socketFile := filepath.Join(emptyDirBasePath, mounterPodSocketFile)
+			if file, err := os.Create(socketFile); err == nil {
+				file.Close()
+			}
+
+			err = ns.mountToNode(ctx, podUID, stagingPath, volumeID)
+			if (err != nil) != tc.wantErr {
+				t.Fatalf("mountToNode() error = %v, wantErr %v", err, tc.wantErr)
+			}
+
+			// Verify that the symlink was removed (because it is deferred inside mountToNode)
+			symlink := filepath.Join(ns.driver.config.FeatureOptions.SharedMountOptions.FuseSocketDir, mounterPodSocketDir, podUID)
+			if _, err := os.Lstat(symlink); !os.IsNotExist(err) {
+				t.Errorf("expected symlink %q to be removed, but it still exists or had another error", symlink)
+			}
+		})
+	}
+}
+
 func validateMountPoint(t *testing.T, fm *mount.FakeMounter, e *mount.MountPoint, unexpectedOpts []string) {
 	t.Helper()
 	if e == nil {

pkg/csi_driver/shared_mount.go

@@ -46,6 +46,7 @@ const (
 	mounterPodMountDir            = "mount-dir"
 	mounterPodSocketFile          = "mounter.sock"
 	mounterPodManagedImageKeyword = "managed"
+	mounterPodSocketDir           = "mount-socket"
 )
 
 var (