support multiple filers

fix https://github.com/seaweedfs/seaweedfs-csi-driver/issues/41

4 files changed, 48 insertions, 12 deletions

diff --git a/go.mod b/go.mod
index d24b449..a6addde 100644
--- a/go.mod
+++ b/go.mod
@@ -5,7 +5,7 @@ go 1.17
 require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.1.1 // indirect
-	github.com/chrislusf/seaweedfs v0.0.0-20210914051334-119d5908dd52
+	github.com/chrislusf/seaweedfs v0.0.0-20211110202745-b0b2a03ea726
 	github.com/container-storage-interface/spec v1.5.0
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/disintegration/imaging v1.6.2 // indirect
@@ -46,12 +46,12 @@ require (
 	golang.org/x/image v0.0.0-20210628002857-a66eb6448b8d // indirect
 	golang.org/x/net v0.0.0-20210813160813-60bc85c4be6d
 	golang.org/x/oauth2 v0.0.0-20210819190943-2bc19b11175f // indirect
-	golang.org/x/sys v0.0.0-20210823070655-63515b42dcdf // indirect
+	golang.org/x/sys v0.0.0-20210908233432-aa78b53d3365 // indirect
 	golang.org/x/term v0.0.0-20210615171337-6886f2dfbf5b // indirect
 	golang.org/x/text v0.3.6 // indirect
 	golang.org/x/time v0.0.0-20210611083556-38a9dc6acbc6 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
-	google.golang.org/genproto v0.0.0-20210909211513-a8c4777a87af // indirect
+	google.golang.org/genproto v0.0.0-20210921142501-181ce0d877f6 // indirect
 	google.golang.org/grpc v1.40.0
 	google.golang.org/protobuf v1.27.1 // indirect
 	gopkg.in/check.v1 v1.0.0-20200902074654-038fdea0a05b // indirect
diff --git a/go.sum b/go.sum
index 1db5c7d..385a816 100644
--- a/go.sum
+++ b/go.sum
@@ -73,6 +73,8 @@ github.com/cespare/xxhash/v2 v2.1.1 h1:6MnRN8NT7+YBpUIWxHtefFZOKTAPgGjpQSxqLNn0+
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/chrislusf/seaweedfs v0.0.0-20210914051334-119d5908dd52 h1:SjO6Z0KyJQFQlCm6wFjnehyaL7U/icwHugmHxeUWX24=
 github.com/chrislusf/seaweedfs v0.0.0-20210914051334-119d5908dd52/go.mod h1:Qt9uvjkTlQG19of3taiwD4jbWVeVr6LYAPuMfJpBKZY=
+github.com/chrislusf/seaweedfs v0.0.0-20211110202745-b0b2a03ea726 h1:v9nhFqEZc21T9ixY2wvhkLwjX8k/uiC+4p77X9r0c6E=
+github.com/chrislusf/seaweedfs v0.0.0-20211110202745-b0b2a03ea726/go.mod h1:8zovCx9OYusQAK7lHAcVFtQsKO2XVQHHvPdlWbK1VRQ=
 github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
 github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
 github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
@@ -578,6 +580,8 @@ golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210823070655-63515b42dcdf h1:2ucpDCmfkl8Bd/FsLtiD653Wf96cW37s+iGx93zsu4k=
 golang.org/x/sys v0.0.0-20210823070655-63515b42dcdf/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20210908233432-aa78b53d3365 h1:6wSTsvPddg9gc/mVEEyk9oOAoxn+bT4Z9q1zx+4RwA4=
+golang.org/x/sys v0.0.0-20210908233432-aa78b53d3365/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
@@ -731,6 +735,8 @@ google.golang.org/genproto v0.0.0-20210402141018-6c239bbf2bb1/go.mod h1:9lPAdzaE
 google.golang.org/genproto v0.0.0-20210602131652-f16073e35f0c/go.mod h1:UODoCrxHCcBojKKwX1terBiRUaqAsFqJiF615XL43r0=
 google.golang.org/genproto v0.0.0-20210909211513-a8c4777a87af h1:aLMMXFYqw01RA6XJim5uaN+afqNNjc9P8HPAbnpnc5s=
 google.golang.org/genproto v0.0.0-20210909211513-a8c4777a87af/go.mod h1:eFjDcFEctNawg4eG61bRv87N7iHBWyVhJu7u1kqDUXY=
+google.golang.org/genproto v0.0.0-20210921142501-181ce0d877f6 h1:2ncG/LajxmrclaZH+ppVi02rQxz4eXYJzGHdFN4Y9UA=
+google.golang.org/genproto v0.0.0-20210921142501-181ce0d877f6/go.mod h1:5CzLGKJ67TSI2B9POpiiyGha0AjJvZIUgRMt1dSmuhc=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
 google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiqXj38=
 google.golang.org/grpc v1.21.1/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM=
diff --git a/pkg/driver/driver.go b/pkg/driver/driver.go
index 9fb08ff..0286a92 100644
--- a/pkg/driver/driver.go
+++ b/pkg/driver/driver.go
@@ -35,7 +35,8 @@ type SeaweedFsDriver struct {
 	vcap  []*csi.VolumeCapability_AccessMode
 	cscap []*csi.ControllerServiceCapability
 
-	filer             pb.ServerAddress
+	filers            []pb.ServerAddress
+	filerIndex        int
 	grpcDialOption    grpc.DialOption
 	ConcurrentWriters int
 	CacheSizeMB       int64
@@ -55,7 +56,7 @@ func NewSeaweedFsDriver(filer, nodeID, endpoint string) *SeaweedFsDriver {
 		nodeID:         nodeID,
 		name:           driverName,
 		version:        version,
-		filer:          pb.ServerAddress(filer),
+		filers:         pb.ServerAddresses(filer).ToAddresses(),
 		grpcDialOption: security.LoadClientTLS(util.GetViper(), "grpc.client"),
 	}
 
@@ -129,10 +130,33 @@ var _ = filer_pb.FilerClient(&SeaweedFsDriver{})
 
 func (d *SeaweedFsDriver) WithFilerClient(fn func(filer_pb.SeaweedFilerClient) error) error {
 
-	return pb.WithCachedGrpcClient(func(grpcConnection *grpc.ClientConn) error {
-		client := filer_pb.NewSeaweedFilerClient(grpcConnection)
-		return fn(client)
-	}, d.filer.ToGrpcAddress(), d.grpcDialOption)
+	return util.Retry("filer grpc", func() error {
+
+		i := d.filerIndex
+		n := len(d.filers)
+		var err error
+		for x := 0; x < n; x++ {
+
+			err = pb.WithCachedGrpcClient(func(grpcConnection *grpc.ClientConn) error {
+				client := filer_pb.NewSeaweedFilerClient(grpcConnection)
+				return fn(client)
+			}, d.filers[i].ToGrpcAddress(), d.grpcDialOption)
+
+			if err != nil {
+				glog.V(0).Infof("WithFilerClient %d %v: %v", x, d.filers[i], err)
+			} else {
+				d.filerIndex = i
+				return nil
+			}
+
+			i++
+			if i >= n {
+				i = 0
+			}
+
+		}
+		return err
+	})
 
 }
 func (d *SeaweedFsDriver) AdjustedUrl(location *filer_pb.Location) string {
diff --git a/pkg/driver/mounter_seaweedfs.go b/pkg/driver/mounter_seaweedfs.go
index a35d702..2bf35fb 100644
--- a/pkg/driver/mounter_seaweedfs.go
+++ b/pkg/driver/mounter_seaweedfs.go
@@ -3,6 +3,7 @@ package driver
 import (
 	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/glog"
+	"strings"
 )
 
 // Implements Mounter
@@ -29,7 +30,12 @@ func newSeaweedFsMounter(path string, collection string, readOnly bool, driver *
 }
 
 func (seaweedFs *seaweedFsMounter) Mount(target string) error {
-	glog.V(0).Infof("mounting %s %s to %s", seaweedFs.driver.filer, seaweedFs.path, target)
+	glog.V(0).Infof("mounting %v %s to %s", seaweedFs.driver.filers, seaweedFs.path, target)
+
+	var filers []string
+	for _, address := range seaweedFs.driver.filers {
+		filers = append(filers, string(address))
+	}
 
 	args := []string{
 		"mount",
@@ -37,7 +43,7 @@ func (seaweedFs *seaweedFsMounter) Mount(target string) error {
 		"-umask=000",
 		fmt.Sprintf("-dir=%s", target),
 		fmt.Sprintf("-collection=%s", seaweedFs.collection),
-		fmt.Sprintf("-filer=%s", seaweedFs.driver.filer),
+		fmt.Sprintf("-filer=%s", strings.Join(filers, ",")),
 		fmt.Sprintf("-filer.path=%s", seaweedFs.path),
 		fmt.Sprintf("-cacheCapacityMB=%d", seaweedFs.driver.CacheSizeMB),
 	}
@@ -75,7 +81,7 @@ func (seaweedFs *seaweedFsMounter) Mount(target string) error {
 
 	err := fuseMount(target, seaweedFsCmd, args)
 	if err != nil {
-		glog.Errorf("mount %s %s to %s: %s", seaweedFs.driver.filer, seaweedFs.path, target, err)
+		glog.Errorf("mount %v %s to %s: %s", seaweedFs.driver.filers, seaweedFs.path, target, err)
 	}
 	return err
 }