add support for redis cluster

fix https://github.com/chrislusf/seaweedfs/issues/705

4 files changed, 179 insertions, 132 deletions

diff --git a/weed/filer2/configuration.go b/weed/filer2/configuration.go
index 3c8a3675b..dac537673 100644
--- a/weed/filer2/configuration.go
+++ b/weed/filer2/configuration.go
@@ -78,6 +78,12 @@ address  = "localhost:6379"
 password = ""
 db = 0
 
+[redis_cluster]
+enabled = false
+addresses = [
+    "localhost:6379",
+]
+
 `
 )
 
diff --git a/weed/filer2/redis/redis_cluster_store.go b/weed/filer2/redis/redis_cluster_store.go
new file mode 100644
index 000000000..02a62567c
--- /dev/null
+++ b/weed/filer2/redis/redis_cluster_store.go
@@ -0,0 +1,31 @@
+package redis
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/filer2"
+	"github.com/go-redis/redis"
+)
+
+func init() {
+	filer2.Stores = append(filer2.Stores, &RedisClusterStore{})
+}
+
+type RedisClusterStore struct {
+	UniversalRedisStore
+}
+
+func (store *RedisClusterStore) GetName() string {
+	return "redis_cluster"
+}
+
+func (store *RedisClusterStore) Initialize(configuration filer2.Configuration) (err error) {
+	return store.initialize(
+		configuration.GetStringSlice("addresses"),
+	)
+}
+
+func (store *RedisClusterStore) initialize(addresses []string) (err error) {
+	store.Client = redis.NewClusterClient(&redis.ClusterOptions{
+		Addrs: addresses,
+	})
+	return
+}
diff --git a/weed/filer2/redis/redis_store.go b/weed/filer2/redis/redis_store.go
index 8bd158d8c..85236a5af 100644
--- a/weed/filer2/redis/redis_store.go
+++ b/weed/filer2/redis/redis_store.go
@@ -1,16 +1,8 @@
 package redis
 
 import (
-	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/filer2"
-	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/go-redis/redis"
-	"sort"
-	"strings"
-)
-
-const (
-	DIR_LIST_MARKER = "\x00"
 )
 
 func init() {
@@ -18,7 +10,7 @@ func init() {
 }
 
 type RedisStore struct {
-	Client *redis.Client
+	UniversalRedisStore
 }
 
 func (store *RedisStore) GetName() string {
@@ -41,126 +33,3 @@ func (store *RedisStore) initialize(hostPort string, password string, database i
 	})
 	return
 }
-
-func (store *RedisStore) InsertEntry(entry *filer2.Entry) (err error) {
-
-	value, err := entry.EncodeAttributesAndChunks()
-	if err != nil {
-		return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err)
-	}
-
-	_, err = store.Client.Set(string(entry.FullPath), value, 0).Result()
-
-	if err != nil {
-		return fmt.Errorf("persisting %s : %v", entry.FullPath, err)
-	}
-
-	dir, name := entry.FullPath.DirAndName()
-	if name != "" {
-		_, err = store.Client.SAdd(genDirectoryListKey(dir), name).Result()
-		if err != nil {
-			return fmt.Errorf("persisting %s in parent dir: %v", entry.FullPath, err)
-		}
-	}
-
-	return nil
-}
-
-func (store *RedisStore) UpdateEntry(entry *filer2.Entry) (err error) {
-
-	return store.InsertEntry(entry)
-}
-
-func (store *RedisStore) FindEntry(fullpath filer2.FullPath) (entry *filer2.Entry, err error) {
-
-	data, err := store.Client.Get(string(fullpath)).Result()
-	if err == redis.Nil {
-		return nil, filer2.ErrNotFound
-	}
-
-	if err != nil {
-		return nil, fmt.Errorf("get %s : %v", entry.FullPath, err)
-	}
-
-	entry = &filer2.Entry{
-		FullPath: fullpath,
-	}
-	err = entry.DecodeAttributesAndChunks([]byte(data))
-	if err != nil {
-		return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err)
-	}
-
-	return entry, nil
-}
-
-func (store *RedisStore) DeleteEntry(fullpath filer2.FullPath) (err error) {
-
-	_, err = store.Client.Del(string(fullpath)).Result()
-
-	if err != nil {
-		return fmt.Errorf("delete %s : %v", fullpath, err)
-	}
-
-	dir, name := fullpath.DirAndName()
-	if name != "" {
-		_, err = store.Client.SRem(genDirectoryListKey(dir), name).Result()
-		if err != nil {
-			return fmt.Errorf("delete %s in parent dir: %v", fullpath, err)
-		}
-	}
-
-	return nil
-}
-
-func (store *RedisStore) ListDirectoryEntries(fullpath filer2.FullPath, startFileName string, inclusive bool,
-	limit int) (entries []*filer2.Entry, err error) {
-
-	members, err := store.Client.SMembers(genDirectoryListKey(string(fullpath))).Result()
-	if err != nil {
-		return nil, fmt.Errorf("list %s : %v", fullpath, err)
-	}
-
-	// skip
-	if startFileName != "" {
-		var t []string
-		for _, m := range members {
-			if strings.Compare(m, startFileName) >= 0 {
-				if m == startFileName {
-					if inclusive {
-						t = append(t, m)
-					}
-				} else {
-					t = append(t, m)
-				}
-			}
-		}
-		members = t
-	}
-
-	// sort
-	sort.Slice(members, func(i, j int) bool {
-		return strings.Compare(members[i], members[j]) < 0
-	})
-
-	// limit
-	if limit < len(members) {
-		members = members[:limit]
-	}
-
-	// fetch entry meta
-	for _, fileName := range members {
-		path := filer2.NewFullPath(string(fullpath), fileName)
-		entry, err := store.FindEntry(path)
-		if err != nil {
-			glog.V(0).Infof("list %s : %v", path, err)
-		} else {
-			entries = append(entries, entry)
-		}
-	}
-
-	return entries, err
-}
-
-func genDirectoryListKey(dir string) (dirList string) {
-	return dir + DIR_LIST_MARKER
-}
diff --git a/weed/filer2/redis/universal_redis_store.go b/weed/filer2/redis/universal_redis_store.go
new file mode 100644
index 000000000..dcf375a57
--- /dev/null
+++ b/weed/filer2/redis/universal_redis_store.go
@@ -0,0 +1,141 @@
+package redis
+
+import (
+	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/filer2"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/go-redis/redis"
+	"sort"
+	"strings"
+)
+
+const (
+	DIR_LIST_MARKER = "\x00"
+)
+
+type UniversalRedisStore struct {
+	Client redis.UniversalClient
+}
+
+func (store *UniversalRedisStore) InsertEntry(entry *filer2.Entry) (err error) {
+
+	value, err := entry.EncodeAttributesAndChunks()
+	if err != nil {
+		return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err)
+	}
+
+	_, err = store.Client.Set(string(entry.FullPath), value, 0).Result()
+
+	if err != nil {
+		return fmt.Errorf("persisting %s : %v", entry.FullPath, err)
+	}
+
+	dir, name := entry.FullPath.DirAndName()
+	if name != "" {
+		_, err = store.Client.SAdd(genDirectoryListKey(dir), name).Result()
+		if err != nil {
+			return fmt.Errorf("persisting %s in parent dir: %v", entry.FullPath, err)
+		}
+	}
+
+	return nil
+}
+
+func (store *UniversalRedisStore) UpdateEntry(entry *filer2.Entry) (err error) {
+
+	return store.InsertEntry(entry)
+}
+
+func (store *UniversalRedisStore) FindEntry(fullpath filer2.FullPath) (entry *filer2.Entry, err error) {
+
+	data, err := store.Client.Get(string(fullpath)).Result()
+	if err == redis.Nil {
+		return nil, filer2.ErrNotFound
+	}
+
+	if err != nil {
+		return nil, fmt.Errorf("get %s : %v", entry.FullPath, err)
+	}
+
+	entry = &filer2.Entry{
+		FullPath: fullpath,
+	}
+	err = entry.DecodeAttributesAndChunks([]byte(data))
+	if err != nil {
+		return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err)
+	}
+
+	return entry, nil
+}
+
+func (store *UniversalRedisStore) DeleteEntry(fullpath filer2.FullPath) (err error) {
+
+	_, err = store.Client.Del(string(fullpath)).Result()
+
+	if err != nil {
+		return fmt.Errorf("delete %s : %v", fullpath, err)
+	}
+
+	dir, name := fullpath.DirAndName()
+	if name != "" {
+		_, err = store.Client.SRem(genDirectoryListKey(dir), name).Result()
+		if err != nil {
+			return fmt.Errorf("delete %s in parent dir: %v", fullpath, err)
+		}
+	}
+
+	return nil
+}
+
+func (store *UniversalRedisStore) ListDirectoryEntries(fullpath filer2.FullPath, startFileName string, inclusive bool,
+	limit int) (entries []*filer2.Entry, err error) {
+
+	members, err := store.Client.SMembers(genDirectoryListKey(string(fullpath))).Result()
+	if err != nil {
+		return nil, fmt.Errorf("list %s : %v", fullpath, err)
+	}
+
+	// skip
+	if startFileName != "" {
+		var t []string
+		for _, m := range members {
+			if strings.Compare(m, startFileName) >= 0 {
+				if m == startFileName {
+					if inclusive {
+						t = append(t, m)
+					}
+				} else {
+					t = append(t, m)
+				}
+			}
+		}
+		members = t
+	}
+
+	// sort
+	sort.Slice(members, func(i, j int) bool {
+		return strings.Compare(members[i], members[j]) < 0
+	})
+
+	// limit
+	if limit < len(members) {
+		members = members[:limit]
+	}
+
+	// fetch entry meta
+	for _, fileName := range members {
+		path := filer2.NewFullPath(string(fullpath), fileName)
+		entry, err := store.FindEntry(path)
+		if err != nil {
+			glog.V(0).Infof("list %s : %v", path, err)
+		} else {
+			entries = append(entries, entry)
+		}
+	}
+
+	return entries, err
+}
+
+func genDirectoryListKey(dir string) (dirList string) {
+	return dir + DIR_LIST_MARKER
+}