Add new option in nodeserver.go to mounting using blobfuse proxy

Author: boddumanohar

Makefile

@@ -31,7 +31,7 @@ BUILD_DATE ?= $(shell date -u +"%Y-%m-%dT%H:%M:%SZ")
 LDFLAGS ?= "-X ${PKG}/pkg/blob.driverVersion=${IMAGE_VERSION} -X ${PKG}/pkg/blob.gitCommit=${GIT_COMMIT} -X ${PKG}/pkg/blob.buildDate=${BUILD_DATE} -s -w -extldflags '-static'"
 E2E_HELM_OPTIONS ?= --set image.blob.pullPolicy=Always --set image.blob.repository=$(REGISTRY)/$(IMAGE_NAME) --set image.blob.tag=$(IMAGE_VERSION)
 ifdef ENABLE_BLOBFUSE_PROXY
-override E2E_HELM_OPTIONS := --set image.blob.pullPolicy=Always --set image.blob.repository=$(REGISTRY)/$(IMAGE_NAME) --set image.blob.tag=$(IMAGE_VERSION) --set controller.logLevel=6 --set node.logLevel=6
+override E2E_HELM_OPTIONS := $(E2E_HELM_OPTIONS) --set controller.logLevel=6 --set node.logLevel=6 --set node.enableBlobfuseProxy=true
 endif
 GINKGO_FLAGS = -ginkgo.v
 GO111MODULE = on
@@ -66,6 +66,9 @@ e2e-test:
 e2e-bootstrap: install-helm install-blobfuse-proxy
 	# Only build and push the image if it does not exist in the registry
 	docker pull $(IMAGE_TAG) || make blob-container push
+	if [[ -z "$(ENABLE_BLOBFUSE_PROXY)" ]]; then \
+		make install-blobfuse-proxy;\
+	fi
 	helm install blob-csi-driver ./charts/latest/blob-csi-driver --namespace kube-system --wait --timeout=15m -v=5 --debug \
 		--set controller.runOnMaster=true \
 		--set controller.replicas=1 \
@@ -168,11 +171,7 @@ blobfuse-proxy-container:
 
 .PHONY: install-blobfuse-proxy
 install-blobfuse-proxy:
-ifdef ENABLE_BLOBFUSE_PROXY
 	kubectl apply -f ./deploy/blobfuse-proxy/blobfuse-proxy.yaml
-else
-	echo "ENABLE_BLOBFUSE_PROXY env not found"
-endif
 
 .PHONY: uninstall-blobfuse-proxy
 uninstall-blobfuse-proxy:

charts/latest/blob-csi-driver-v1.1.0.tgz

@@ -85,6 +85,8 @@ spec:
           args:
             - "--v={{ .Values.node.logLevel }}"
             - "--endpoint=$(CSI_ENDPOINT)"
+            - "--blobfuse-proxy-endpoint=$(BLOBFUSE_PROXY_ENDPOINT)"
+            - "--enable-blobfuse-proxy={{ .Values.node.enableBlobfuseProxy }}"
             - "--nodeid=$(KUBE_NODE_NAME)"
             - "--metrics-address=0.0.0.0:{{ .Values.node.metricsPort }}"
           ports:
@@ -111,6 +113,8 @@ spec:
                   optional: true
             - name: CSI_ENDPOINT
               value: unix:///csi/csi.sock
+            - name: BLOBFUSE_PROXY_ENDPOINT
+              value: unix:///csi/blobfuse-proxy.sock
             - name: KUBE_NODE_NAME
               valueFrom:
                 fieldRef:

charts/latest/blob-csi-driver/templates/csi-blob-node.yaml

@@ -40,6 +40,7 @@ controller:
 node:
   metricsPort: 29635
   logLevel: 5
+  enableBlobfuseProxy: false
 
 kubelet:
   linuxPath: /var/lib/kubelet

charts/latest/blob-csi-driver/values.yaml

@@ -81,6 +81,7 @@ spec:
           args:
             - "--v=5"
             - "--endpoint=$(CSI_ENDPOINT)"
+            - "--blobfuse-proxy-endpoint=$(BLOBFUSE_PROXY_ENDPOINT)"
             - "--nodeid=$(KUBE_NODE_NAME)"
             - "--metrics-address=0.0.0.0:29635"
           ports:
@@ -107,6 +108,8 @@ spec:
                   optional: true
             - name: CSI_ENDPOINT
               value: unix:///csi/csi.sock
+            - name: BLOBFUSE_PROXY_ENDPOINT
+              value: unix:///csi/blobfuse-proxy.sock
             - name: KUBE_NODE_NAME
               valueFrom:
                 fieldRef:

deploy/csi-blob-node.yaml

@@ -93,23 +93,29 @@ var (
 // Driver implements all interfaces of CSI drivers
 type Driver struct {
 	csicommon.CSIDriver
-	cloud      *azure.Cloud
-	mounter    *mount.SafeFormatAndMount
-	volLockMap *util.LockMap
+	cloud                   *azure.Cloud
+	blobfuseProxyEndpoint   string
+	enableBlobfuseProxy     bool
+	blobfuseProxyConnTimout int
+	mounter                 *mount.SafeFormatAndMount
+	volLockMap              *util.LockMap
 	// A map storing all volumes with ongoing operations so that additional operations
 	// for that same volume (as defined by VolumeID) return an Aborted error
 	volumeLocks *volumeLocks
 }
 
 // NewDriver Creates a NewCSIDriver object. Assumes vendor version is equal to driver version &
 // does not support optional driver plugin info manifest field. Refer to CSI spec for more details.
-func NewDriver(nodeID string) *Driver {
+func NewDriver(nodeID, blobfuseProxyEndpoint string, enableBlobfuseProxy bool, blobfuseProxyConnTimout int) *Driver {
 	driver := Driver{}
 	driver.Name = DriverName
 	driver.Version = driverVersion
 	driver.NodeID = nodeID
 	driver.volLockMap = util.NewLockMap()
 	driver.volumeLocks = newVolumeLocks()
+	driver.blobfuseProxyEndpoint = blobfuseProxyEndpoint
+	driver.enableBlobfuseProxy = enableBlobfuseProxy
+	driver.blobfuseProxyConnTimout = blobfuseProxyConnTimout
 	return &driver
 }
 

pkg/blob/blob.go

@@ -44,19 +44,19 @@ const (
 )
 
 func NewFakeDriver() *Driver {
-	driver := NewDriver(fakeNodeID)
+	driver := NewDriver(fakeNodeID, "", false, 5)
 	driver.Name = fakeDriverName
 	driver.Version = vendorVersion
 	return driver
 }
 
 func TestNewFakeDriver(t *testing.T) {
-	d := NewDriver(fakeNodeID)
+	d := NewDriver(fakeNodeID, "", false, 5)
 	assert.NotNil(t, d)
 }
 
 func TestNewDriver(t *testing.T) {
-	driver := NewDriver(fakeNodeID)
+	driver := NewDriver(fakeNodeID, "", false, 5)
 	fakedriver := NewFakeDriver()
 	fakedriver.Name = DriverName
 	fakedriver.Version = driverVersion

pkg/blob/blob_test.go

@@ -38,8 +38,20 @@ import (
 	"google.golang.org/grpc/status"
 
 	"golang.org/x/net/context"
+	"google.golang.org/grpc"
+	mount_azure_blob "sigs.k8s.io/blob-csi-driver/pkg/blobfuse-proxy/pb"
 )
 
+type MountClient struct {
+	service mount_azure_blob.MountServiceClient
+}
+
+// NewMountClient returns a new mount client
+func NewMountClient(cc *grpc.ClientConn) *MountClient {
+	service := mount_azure_blob.NewMountServiceClient(cc)
+	return &MountClient{service}
+}
+
 // NodePublishVolume mount the volume from staging to target path
 func (d *Driver) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) {
 	if req.GetVolumeCapability() == nil {
@@ -91,6 +103,38 @@ func (d *Driver) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolu
 	return &csi.NodePublishVolumeResponse{}, nil
 }
 
+func (d *Driver) mountBlobfuseWithProxy(args string, authEnv []string) (string, error) {
+	klog.V(2).Infof("mouting using blobfuse proxy")
+	var resp *mount_azure_blob.MountAzureBlobResponse
+	var output string
+	connectionTimout := time.Duration(d.blobfuseProxyConnTimout)
+	ctx, cancel := context.WithTimeout(context.Background(), connectionTimout*time.Second)
+	defer cancel()
+	conn, err := grpc.DialContext(ctx, d.blobfuseProxyEndpoint, grpc.WithInsecure(), grpc.WithBlock())
+	if err == nil {
+		mountClient := NewMountClient(conn)
+		mountreq := mount_azure_blob.MountAzureBlobRequest{
+			MountArgs: args,
+			AuthEnv:   authEnv,
+		}
+		klog.V(2).Infof("calling BlobfuseProxy: MountAzureBlob function")
+		resp, err = mountClient.service.MountAzureBlob(context.TODO(), &mountreq)
+		if err != nil {
+			klog.Error("GRPC call returned with an error:", err)
+		}
+		output = resp.GetOutput()
+	}
+	return output, err
+}
+
+func (d *Driver) mountBlobfuseInsideDriver(args string, authEnv []string) (string, error) {
+	klog.V(2).Infof("mounting blobfuse inside driver")
+	cmd := exec.Command("blobfuse", strings.Split(args, " ")...)
+	cmd.Env = append(os.Environ(), authEnv...)
+	output, err := cmd.CombinedOutput()
+	return string(output), err
+}
+
 // NodeUnpublishVolume unmount the volume from the target path
 func (d *Driver) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpublishVolumeRequest) (*csi.NodeUnpublishVolumeResponse, error) {
 	if len(req.GetVolumeId()) == 0 {
@@ -213,14 +257,17 @@ func (d *Driver) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRe
 
 	klog.V(2).Infof("target %v\nprotocol %v\n\nvolumeId %v\ncontext %v\nmountflags %v\nmountOptions %v\nargs %v\nserverAddress %v",
 		targetPath, protocol, volumeID, attrib, mountFlags, mountOptions, args, serverAddress)
-	cmd := exec.Command("blobfuse", strings.Split(args, " ")...)
 
 	authEnv = append(authEnv, "AZURE_STORAGE_ACCOUNT="+accountName, "AZURE_STORAGE_BLOB_ENDPOINT="+serverAddress)
-	cmd.Env = append(os.Environ(), authEnv...)
+	var output string
+	if d.enableBlobfuseProxy {
+		output, err = d.mountBlobfuseWithProxy(args, authEnv)
+	} else {
+		output, err = d.mountBlobfuseInsideDriver(args, authEnv)
+	}
 
-	output, err := cmd.CombinedOutput()
 	if err != nil {
-		err = fmt.Errorf("Mount failed with error: %v, output: %v", err, string(output))
+		err = fmt.Errorf("Mount failed with error: %v, output: %v", err, output)
 		klog.Errorf("%v", err)
 		notMnt, mntErr := d.mounter.IsLikelyNotMountPoint(targetPath)
 		if mntErr != nil {

pkg/blob/nodeserver.go

@@ -26,6 +26,7 @@ import (
 	"syscall"
 	"testing"
 
+	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
@@ -132,6 +133,16 @@ func TestEnsureMountPoint(t *testing.T) {
 	assert.NoError(t, err)
 }
 
+func TestNewMountClient(t *testing.T) {
+	conn, _ := grpc.DialContext(context.TODO(), "unix://tmp/proxy.sock", grpc.WithInsecure())
+	// No need to check for error because the socket is not available and
+	// will result in following error: failed to build resolver: invalid (non-empty) authority: tmp
+	client := NewMountClient(conn)
+	valueType := reflect.TypeOf(client).String()
+	fmt.Println(valueType)
+	assert.Equal(t, valueType, "*blob.MountClient")
+}
+
 func TestNodePublishVolume(t *testing.T) {
 	volumeCap := csi.VolumeCapability_AccessMode{Mode: csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER}
 	createDirError := status.Errorf(codes.Internal,
@@ -559,3 +570,21 @@ func TestNodeExpandVolume(t *testing.T) {
 		t.Errorf("Unexpected error: %v", err)
 	}
 }
+
+func TestMountBlobfuseWithProxy(t *testing.T) {
+	args := "--tmp-path /tmp"
+	authEnv := []string{"username=blob", "authkey=blob"}
+	d := NewFakeDriver()
+	_, err := d.mountBlobfuseWithProxy(args, authEnv)
+	// should be context.deadlineExceededError{} error
+	assert.NotNil(t, err)
+}
+
+func TestMountBlobfuseInsideDriver(t *testing.T) {
+	args := "--tmp-path /tmp"
+	authEnv := []string{"username=blob", "authkey=blob"}
+	d := NewFakeDriver()
+	_, err := d.mountBlobfuseInsideDriver(args, authEnv)
+	// the error should be of type exec.ExitError
+	assert.NotNil(t, err)
+}

pkg/blob/nodeserver_test.go

@@ -36,11 +36,14 @@ func init() {
 }
 
 var (
-	endpoint       = flag.String("endpoint", "unix://tmp/csi.sock", "CSI endpoint")
-	nodeID         = flag.String("nodeid", "", "node id")
-	version        = flag.Bool("version", false, "Print the version and exit.")
-	metricsAddress = flag.String("metrics-address", "0.0.0.0:29634", "export the metrics")
-	kubeconfig     = flag.String("kubeconfig", "", "Absolute path to the kubeconfig file. Required only when running out of cluster.")
+	endpoint                = flag.String("endpoint", "unix://tmp/csi.sock", "CSI endpoint")
+	blobfuseProxyEndpoint   = flag.String("blobfuse-proxy-endpoint", "unix://tmp/blobfuse-proxy.sock", "blobfuse-proxy endpoint")
+	nodeID                  = flag.String("nodeid", "", "node id")
+	version                 = flag.Bool("version", false, "Print the version and exit.")
+	metricsAddress          = flag.String("metrics-address", "0.0.0.0:29634", "export the metrics")
+	kubeconfig              = flag.String("kubeconfig", "", "Absolute path to the kubeconfig file. Required only when running out of cluster.")
+	enableBlobfuseProxy     = flag.Bool("enable-blobfuse-proxy", false, "Whether supports using Blobfuse proxy for mounts")
+	blobfuseProxyConnTimout = flag.Int("blobfuse-proxy-connect-timeout", 5, "blobfuse proxy connection timeout(seconds)")
 )
 
 func main() {
@@ -61,7 +64,7 @@ func main() {
 }
 
 func handle() {
-	driver := blob.NewDriver(*nodeID)
+	driver := blob.NewDriver(*nodeID, *blobfuseProxyEndpoint, *enableBlobfuseProxy, *blobfuseProxyConnTimout)
 	if driver == nil {
 		klog.Fatalln("Failed to initialize Azure Blob Storage CSI driver")
 	}