filer store: add redis2 redis_cluster2 as default redis store

5 files changed, 246 insertions, 2 deletions

diff --git a/weed/command/scaffold.go b/weed/command/scaffold.go
index 109f0c04f..67fa8dcf3 100644
--- a/weed/command/scaffold.go
+++ b/weed/command/scaffold.go
@@ -143,13 +143,13 @@ hosts=[
 	"localhost:9042",
 ]
 
-[redis]
+[redis2]
 enabled = false
 address  = "localhost:6379"
 password = ""
 database = 0
 
-[redis_cluster]
+[redis_cluster2]
 enabled = false
 addresses = [
     "localhost:30001",
diff --git a/weed/filer2/redis2/redis_cluster_store.go b/weed/filer2/redis2/redis_cluster_store.go
new file mode 100644
index 000000000..b252eabab
--- /dev/null
+++ b/weed/filer2/redis2/redis_cluster_store.go
@@ -0,0 +1,42 @@
+package redis2
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/filer2"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/go-redis/redis"
+)
+
+func init() {
+	filer2.Stores = append(filer2.Stores, &RedisCluster2Store{})
+}
+
+type RedisCluster2Store struct {
+	UniversalRedis2Store
+}
+
+func (store *RedisCluster2Store) GetName() string {
+	return "redis_cluster2"
+}
+
+func (store *RedisCluster2Store) Initialize(configuration util.Configuration, prefix string) (err error) {
+
+	configuration.SetDefault(prefix+"useReadOnly", true)
+	configuration.SetDefault(prefix+"routeByLatency", true)
+
+	return store.initialize(
+		configuration.GetStringSlice(prefix+"addresses"),
+		configuration.GetString(prefix+"password"),
+		configuration.GetBool(prefix+"useReadOnly"),
+		configuration.GetBool(prefix+"routeByLatency"),
+	)
+}
+
+func (store *RedisCluster2Store) initialize(addresses []string, password string, readOnly, routeByLatency bool) (err error) {
+	store.Client = redis.NewClusterClient(&redis.ClusterOptions{
+		Addrs:          addresses,
+		Password:       password,
+		ReadOnly:       readOnly,
+		RouteByLatency: routeByLatency,
+	})
+	return
+}
diff --git a/weed/filer2/redis2/redis_store.go b/weed/filer2/redis2/redis_store.go
new file mode 100644
index 000000000..1e2a20043
--- /dev/null
+++ b/weed/filer2/redis2/redis_store.go
@@ -0,0 +1,36 @@
+package redis2
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/filer2"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/go-redis/redis"
+)
+
+func init() {
+	filer2.Stores = append(filer2.Stores, &Redis2Store{})
+}
+
+type Redis2Store struct {
+	UniversalRedis2Store
+}
+
+func (store *Redis2Store) GetName() string {
+	return "redis2"
+}
+
+func (store *Redis2Store) Initialize(configuration util.Configuration, prefix string) (err error) {
+	return store.initialize(
+		configuration.GetString(prefix+"address"),
+		configuration.GetString(prefix+"password"),
+		configuration.GetInt(prefix+"database"),
+	)
+}
+
+func (store *Redis2Store) initialize(hostPort string, password string, database int) (err error) {
+	store.Client = redis.NewClient(&redis.Options{
+		Addr:     hostPort,
+		Password: password,
+		DB:       database,
+	})
+	return
+}
diff --git a/weed/filer2/redis2/universal_redis_store.go b/weed/filer2/redis2/universal_redis_store.go
new file mode 100644
index 000000000..457ab50c6
--- /dev/null
+++ b/weed/filer2/redis2/universal_redis_store.go
@@ -0,0 +1,165 @@
+package redis2
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/go-redis/redis"
+
+	"github.com/chrislusf/seaweedfs/weed/filer2"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/util"
+)
+
+const (
+	DIR_LIST_MARKER = "\x00"
+)
+
+type UniversalRedis2Store struct {
+	Client redis.UniversalClient
+}
+
+func (store *UniversalRedis2Store) BeginTransaction(ctx context.Context) (context.Context, error) {
+	return ctx, nil
+}
+func (store *UniversalRedis2Store) CommitTransaction(ctx context.Context) error {
+	return nil
+}
+func (store *UniversalRedis2Store) RollbackTransaction(ctx context.Context) error {
+	return nil
+}
+
+func (store *UniversalRedis2Store) InsertEntry(ctx context.Context, 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, time.Duration(entry.TtlSec)*time.Second).Result()
+
+	if err != nil {
+		return fmt.Errorf("persisting %s : %v", entry.FullPath, err)
+	}
+
+	dir, name := entry.FullPath.DirAndName()
+	if name != "" {
+		_, err = store.Client.ZAdd(genDirectoryListKey(dir), &redis.Z{Score: 0, Member: name}).Result()
+		if err != nil {
+			return fmt.Errorf("persisting %s in parent dir: %v", entry.FullPath, err)
+		}
+	}
+
+	return nil
+}
+
+func (store *UniversalRedis2Store) UpdateEntry(ctx context.Context, entry *filer2.Entry) (err error) {
+
+	return store.InsertEntry(ctx, entry)
+}
+
+func (store *UniversalRedis2Store) FindEntry(ctx context.Context, fullpath util.FullPath) (entry *filer2.Entry, err error) {
+
+	data, err := store.Client.Get(string(fullpath)).Result()
+	if err == redis.Nil {
+		return nil, filer_pb.ErrNotFound
+	}
+
+	if err != nil {
+		return nil, fmt.Errorf("get %s : %v", 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 *UniversalRedis2Store) DeleteEntry(ctx context.Context, fullpath util.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.ZRem(genDirectoryListKey(dir), name).Result()
+		if err != nil {
+			return fmt.Errorf("delete %s in parent dir: %v", fullpath, err)
+		}
+	}
+
+	return nil
+}
+
+func (store *UniversalRedis2Store) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) (err error) {
+
+	members, err := store.Client.ZRange(genDirectoryListKey(string(fullpath)), 0, -1).Result()
+	if err != nil {
+		return fmt.Errorf("delete folder %s : %v", fullpath, err)
+	}
+
+	for _, fileName := range members {
+		path := util.NewFullPath(string(fullpath), fileName)
+		_, err = store.Client.Del(string(path)).Result()
+		if err != nil {
+			return fmt.Errorf("delete %s in parent dir: %v", fullpath, err)
+		}
+	}
+
+	return nil
+}
+
+func (store *UniversalRedis2Store) ListDirectoryEntries(ctx context.Context, fullpath util.FullPath, startFileName string, inclusive bool,
+	limit int) (entries []*filer2.Entry, err error) {
+
+	dirListKey := genDirectoryListKey(string(fullpath))
+	start := int64(0)
+	if startFileName != "" {
+		start, _ = store.Client.ZRank(dirListKey, startFileName).Result()
+		if !inclusive {
+			start++
+		}
+	}
+	members, err := store.Client.ZRange(dirListKey, start, start+int64(limit)-1).Result()
+	if err != nil {
+		return nil, fmt.Errorf("list %s : %v", fullpath, err)
+	}
+
+	// fetch entry meta
+	for _, fileName := range members {
+		path := util.NewFullPath(string(fullpath), fileName)
+		entry, err := store.FindEntry(ctx, path)
+		if err != nil {
+			glog.V(0).Infof("list %s : %v", path, err)
+		} else {
+			if entry.TtlSec > 0 {
+				if entry.Attr.Crtime.Add(time.Duration(entry.TtlSec) * time.Second).Before(time.Now()) {
+					store.Client.Del(string(path)).Result()
+					store.Client.ZRem(dirListKey, fileName).Result()
+					continue
+				}
+			}
+			entries = append(entries, entry)
+		}
+	}
+
+	return entries, err
+}
+
+func genDirectoryListKey(dir string) (dirList string) {
+	return dir + DIR_LIST_MARKER
+}
+
+func (store *UniversalRedis2Store) Shutdown() {
+	store.Client.Close()
+}
diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go
index cf8669202..63bc8bd03 100644
--- a/weed/server/filer_server.go
+++ b/weed/server/filer_server.go
@@ -24,6 +24,7 @@ import (
 	_ "github.com/chrislusf/seaweedfs/weed/filer2/mysql"
 	_ "github.com/chrislusf/seaweedfs/weed/filer2/postgres"
 	_ "github.com/chrislusf/seaweedfs/weed/filer2/redis"
+	_ "github.com/chrislusf/seaweedfs/weed/filer2/redis2"
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/notification"
 	_ "github.com/chrislusf/seaweedfs/weed/notification/aws_sqs"