support expand volume

5 files changed, 59 insertions, 18 deletions

diff --git a/pkg/driver/controllerserver.go b/pkg/driver/controllerserver.go
index d8be023..6f57359 100644
--- a/pkg/driver/controllerserver.go
+++ b/pkg/driver/controllerserver.go
@@ -5,14 +5,17 @@ import (
 	"crypto/sha1"
 	"encoding/hex"
 	"fmt"
-	"io"
-	"strings"
-
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/mount_pb"
 	"github.com/container-storage-interface/spec/lib/go/csi"
+	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	_ "google.golang.org/grpc/resolver/passthrough"
 	"google.golang.org/grpc/status"
+	"io"
+	"strings"
 )
 
 type ControllerServer struct {
@@ -165,7 +168,22 @@ func (cs *ControllerServer) ListSnapshots(ctx context.Context, req *csi.ListSnap
 }
 
 func (cs *ControllerServer) ControllerExpandVolume(ctx context.Context, req *csi.ControllerExpandVolumeRequest) (*csi.ControllerExpandVolumeResponse, error) {
-	return nil, status.Error(codes.Unimplemented, "")
+
+	clientConn, err := grpc.Dial("passthrough:///unix://"+cs.Driver.mountSocket, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		return nil, err
+	}
+	defer clientConn.Close()
+
+	client := mount_pb.NewSeaweedMountClient(clientConn)
+	_, err = client.Configure(context.Background(), &mount_pb.ConfigureRequest{
+		CollectionCapacity: req.CapacityRange.LimitBytes,
+	})
+
+	return &csi.ControllerExpandVolumeResponse{
+		CapacityBytes: req.CapacityRange.LimitBytes,
+	}, err
+
 }
 
 func (cs *ControllerServer) ControllerGetVolume(ctx context.Context, req *csi.ControllerGetVolumeRequest) (*csi.ControllerGetVolumeResponse, error) {
diff --git a/pkg/driver/driver.go b/pkg/driver/driver.go
index 5e82a60..00e1f35 100644
--- a/pkg/driver/driver.go
+++ b/pkg/driver/driver.go
@@ -30,7 +30,8 @@ type SeaweedFsDriver struct {
 	nodeID  string
 	version string
 
-	endpoint string
+	endpoint    string
+	mountSocket string
 
 	vcap  []*csi.VolumeCapability_AccessMode
 	cscap []*csi.ControllerServiceCapability
@@ -53,6 +54,12 @@ func NewSeaweedFsDriver(filer, nodeID, endpoint string) *SeaweedFsDriver {
 
 	util.LoadConfiguration("security", false)
 
+	montDirHash := util.HashToInt32([]byte(endpoint))
+	if montDirHash < 0 {
+		montDirHash = -montDirHash
+	}
+	socket := fmt.Sprintf("/tmp/seaweedfs-mount-%d.sock", montDirHash)
+
 	n := &SeaweedFsDriver{
 		endpoint:       endpoint,
 		nodeID:         nodeID,
@@ -60,6 +67,7 @@ func NewSeaweedFsDriver(filer, nodeID, endpoint string) *SeaweedFsDriver {
 		version:        version,
 		filers:         pb.ServerAddresses(filer).ToAddresses(),
 		grpcDialOption: security.LoadClientTLS(util.GetViper(), "grpc.client"),
+		mountSocket:    socket,
 	}
 
 	n.AddVolumeCapabilityAccessModes([]csi.VolumeCapability_AccessMode_Mode{
diff --git a/pkg/driver/identityserver.go b/pkg/driver/identityserver.go
index 7d0fab7..55ce934 100644
--- a/pkg/driver/identityserver.go
+++ b/pkg/driver/identityserver.go
@@ -1,8 +1,8 @@
 package driver
 
 import (
-	"github.com/container-storage-interface/spec/lib/go/csi"
 	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/container-storage-interface/spec/lib/go/csi"
 	"golang.org/x/net/context"
 )
 
@@ -35,15 +35,13 @@ func (ids *IdentityServer) GetPluginCapabilities(ctx context.Context, req *csi.G
 					},
 				},
 			},
-			/* // TODO add later
-				{
-					Type: &csi.PluginCapability_VolumeExpansion_{
-						VolumeExpansion: &csi.PluginCapability_VolumeExpansion{
-							Type: csi.PluginCapability_VolumeExpansion_ONLINE,
-						},
+			{
+				Type: &csi.PluginCapability_VolumeExpansion_{
+					VolumeExpansion: &csi.PluginCapability_VolumeExpansion{
+						Type: csi.PluginCapability_VolumeExpansion_ONLINE,
 					},
 				},
-			*/
+			},
 		},
 	}, nil
 }
diff --git a/pkg/driver/mounter_seaweedfs.go b/pkg/driver/mounter_seaweedfs.go
index 2d0eba9..b38fe31 100644
--- a/pkg/driver/mounter_seaweedfs.go
+++ b/pkg/driver/mounter_seaweedfs.go
@@ -49,6 +49,7 @@ func (seaweedFs *seaweedFsMounter) Mount(target string) error {
 		fmt.Sprintf("-filer=%s", strings.Join(filers, ",")),
 		fmt.Sprintf("-filer.path=%s", seaweedFs.path),
 		fmt.Sprintf("-cacheCapacityMB=%d", seaweedFs.driver.CacheSizeMB),
+		fmt.Sprintf("-localSocket=%s", seaweedFs.driver.mountSocket),
 	}
 
 	// came from https://github.com/seaweedfs/seaweedfs-csi-driver/pull/12
diff --git a/pkg/driver/nodeserver.go b/pkg/driver/nodeserver.go
index 968f5e4..365fb4c 100644
--- a/pkg/driver/nodeserver.go
+++ b/pkg/driver/nodeserver.go
@@ -3,15 +3,17 @@ package driver
 import (
 	"context"
 	"fmt"
-	"os"
-	"strings"
-
 	"github.com/chrislusf/seaweedfs/weed/glog"
-
+	"github.com/chrislusf/seaweedfs/weed/pb/mount_pb"
 	"github.com/container-storage-interface/spec/lib/go/csi"
+	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	_ "google.golang.org/grpc/resolver/passthrough"
 	"google.golang.org/grpc/status"
 	"k8s.io/utils/mount"
+	"os"
+	"strings"
 )
 
 type NodeServer struct {
@@ -159,7 +161,21 @@ func (ns *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVol
 }
 
 func (ns *NodeServer) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandVolumeRequest) (*csi.NodeExpandVolumeResponse, error) {
-	return &csi.NodeExpandVolumeResponse{}, status.Error(codes.Unimplemented, "NodeExpandVolume is not implemented")
+
+	clientConn, err := grpc.Dial("passthrough:///unix://"+ns.Driver.mountSocket, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		return nil, err
+	}
+	defer clientConn.Close()
+
+	client := mount_pb.NewSeaweedMountClient(clientConn)
+	_, err = client.Configure(context.Background(), &mount_pb.ConfigureRequest{
+		CollectionCapacity: req.CapacityRange.LimitBytes,
+	})
+
+	return &csi.NodeExpandVolumeResponse{
+		CapacityBytes: req.CapacityRange.LimitBytes,
+	}, err
 }
 
 func checkMount(targetPath string) (bool, error) {