FilerStore: add redis_lua

weed/command/scaffold/filer.toml

@@ -195,6 +195,40 @@ routeByLatency = false
 # This changes the data layout. Only add new directories. Removing/Updating will cause data loss.
 superLargeDirectories = []
 
+[redis_lua]
+enabled = false
+address = "localhost:6379"
+password = ""
+database = 0
+# This changes the data layout. Only add new directories. Removing/Updating will cause data loss.
+superLargeDirectories = []
+
+[redis_lua_sentinel]
+enabled = false
+addresses = ["172.22.12.7:26379","172.22.12.8:26379","172.22.12.9:26379"]
+masterName = "master"
+username = ""
+password = ""
+database = 0
+
+[redis_lua_cluster]
+enabled = false
+addresses = [
+    "localhost:30001",
+    "localhost:30002",
+    "localhost:30003",
+    "localhost:30004",
+    "localhost:30005",
+    "localhost:30006",
+]
+password = ""
+# allows reads from slave servers or the master, but all writes still go to the master
+readOnly = false
+# automatically use the closest Redis server for reads
+routeByLatency = false
+# This changes the data layout. Only add new directories. Removing/Updating will cause data loss.
+superLargeDirectories = []
+
 [redis3] # beta
 enabled = false
 address = "localhost:6379"

weed/filer/redis_lua/redis_cluster_store.go

@@ -0,0 +1,44 @@
+package redis_lua
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/go-redis/redis/v8"
+)
+
+func init() {
+	filer.Stores = append(filer.Stores, &RedisLuaClusterStore{})
+}
+
+type RedisLuaClusterStore struct {
+	UniversalRedisLuaStore
+}
+
+func (store *RedisLuaClusterStore) GetName() string {
+	return "redis_lua_cluster"
+}
+
+func (store *RedisLuaClusterStore) Initialize(configuration util.Configuration, prefix string) (err error) {
+
+	configuration.SetDefault(prefix+"useReadOnly", false)
+	configuration.SetDefault(prefix+"routeByLatency", false)
+
+	return store.initialize(
+		configuration.GetStringSlice(prefix+"addresses"),
+		configuration.GetString(prefix+"password"),
+		configuration.GetBool(prefix+"useReadOnly"),
+		configuration.GetBool(prefix+"routeByLatency"),
+		configuration.GetStringSlice(prefix+"superLargeDirectories"),
+	)
+}
+
+func (store *RedisLuaClusterStore) initialize(addresses []string, password string, readOnly, routeByLatency bool, superLargeDirectories []string) (err error) {
+	store.Client = redis.NewClusterClient(&redis.ClusterOptions{
+		Addrs:          addresses,
+		Password:       password,
+		ReadOnly:       readOnly,
+		RouteByLatency: routeByLatency,
+	})
+	store.loadSuperLargeDirectories(superLargeDirectories)
+	return
+}

weed/filer/redis_lua/redis_sentinel_store.go

@@ -0,0 +1,45 @@
+package redis_lua
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/go-redis/redis/v8"
+	"time"
+)
+
+func init() {
+	filer.Stores = append(filer.Stores, &RedisLuaSentinelStore{})
+}
+
+type RedisLuaSentinelStore struct {
+	UniversalRedisLuaStore
+}
+
+func (store *RedisLuaSentinelStore) GetName() string {
+	return "redis_lua_sentinel"
+}
+
+func (store *RedisLuaSentinelStore) Initialize(configuration util.Configuration, prefix string) (err error) {
+	return store.initialize(
+		configuration.GetStringSlice(prefix+"addresses"),
+		configuration.GetString(prefix+"masterName"),
+		configuration.GetString(prefix+"username"),
+		configuration.GetString(prefix+"password"),
+		configuration.GetInt(prefix+"database"),
+	)
+}
+
+func (store *RedisLuaSentinelStore) initialize(addresses []string, masterName string, username string, password string, database int) (err error) {
+	store.Client = redis.NewFailoverClient(&redis.FailoverOptions{
+		MasterName:      masterName,
+		SentinelAddrs:   addresses,
+		Username:        username,
+		Password:        password,
+		DB:              database,
+		MinRetryBackoff: time.Millisecond * 100,
+		MaxRetryBackoff: time.Minute * 1,
+		ReadTimeout:     time.Second * 30,
+		WriteTimeout:    time.Second * 5,
+	})
+	return
+}

weed/filer/redis_lua/redis_store.go

@@ -0,0 +1,38 @@
+package redis_lua
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/go-redis/redis/v8"
+)
+
+func init() {
+	filer.Stores = append(filer.Stores, &RedisLuaStore{})
+}
+
+type RedisLuaStore struct {
+	UniversalRedisLuaStore
+}
+
+func (store *RedisLuaStore) GetName() string {
+	return "redis_lua"
+}
+
+func (store *RedisLuaStore) Initialize(configuration util.Configuration, prefix string) (err error) {
+	return store.initialize(
+		configuration.GetString(prefix+"address"),
+		configuration.GetString(prefix+"password"),
+		configuration.GetInt(prefix+"database"),
+		configuration.GetStringSlice(prefix+"superLargeDirectories"),
+	)
+}
+
+func (store *RedisLuaStore) initialize(hostPort string, password string, database int, superLargeDirectories []string) (err error) {
+	store.Client = redis.NewClient(&redis.Options{
+		Addr:     hostPort,
+		Password: password,
+		DB:       database,
+	})
+	store.loadSuperLargeDirectories(superLargeDirectories)
+	return
+}

weed/filer/redis_lua/stored_procedure/delete_entry.lua

@@ -0,0 +1,19 @@
+-- KEYS[1]: full path of entry
+local fullpath = KEYS[1]
+-- KEYS[2]: full path of entry
+local fullpath_list_key = KEYS[2]
+-- KEYS[3]: dir of the entry
+local dir_list_key = KEYS[3]
+
+-- ARGV[1]: isSuperLargeDirectory
+local isSuperLargeDirectory = ARGV[1] == "1"
+-- ARGV[2]: name of the entry
+local name = ARGV[2]
+
+redis.call("DEL", fullpath, fullpath_list_key)
+
+if not isSuperLargeDirectory and name ~= "" then
+    redis.call("ZREM", dir_list_key, name)
+end
+
+return 0

weed/filer/redis_lua/stored_procedure/delete_folder_children.lua

@@ -0,0 +1,15 @@
+-- KEYS[1]: full path of entry
+local fullpath = KEYS[1]
+
+if fullpath ~= "" and string.sub(fullpath, -1) == "/" then
+    fullpath = string.sub(fullpath, 0, -2)
+end
+
+local files = redis.call("ZRANGE", fullpath .. "\0", "0", "-1")
+
+for _, name in ipairs(files) do
+    local file_path = fullpath .. "/" .. name
+    redis.call("DEL", file_path, file_path .. "\0")
+end
+
+return 0

weed/filer/redis_lua/stored_procedure/init.go

@@ -0,0 +1,24 @@
+package stored_procedure
+
+import (
+	_ "embed"
+	"github.com/go-redis/redis/v8"
+)
+
+func init() {
+	InsertEntryScript = redis.NewScript(insertEntry)
+	DeleteEntryScript = redis.NewScript(deleteEntry)
+	DeleteFolderChildrenScript = redis.NewScript(deleteFolderChildren)
+}
+
+//go:embed insert_entry.lua
+var insertEntry string
+var InsertEntryScript *redis.Script
+
+//go:embed delete_entry.lua
+var deleteEntry string
+var DeleteEntryScript *redis.Script
+
+//go:embed delete_folder_children.lua
+var deleteFolderChildren string
+var DeleteFolderChildrenScript *redis.Script

weed/filer/redis_lua/stored_procedure/insert_entry.lua

@@ -0,0 +1,27 @@
+-- KEYS[1]: full path of entry
+local full_path = KEYS[1]
+-- KEYS[2]: dir of the entry
+local dir_list_key = KEYS[2]
+
+-- ARGV[1]: content of the entry
+local entry = ARGV[1]
+-- ARGV[2]: TTL of the entry
+local ttlSec = tonumber(ARGV[2])
+-- ARGV[3]: isSuperLargeDirectory
+local isSuperLargeDirectory = ARGV[3] == "1"
+-- ARGV[4]: zscore of the entry in zset
+local zscore = tonumber(ARGV[4])
+-- ARGV[5]: name of the entry
+local name = ARGV[5]
+
+if ttlSec > 0 then
+    redis.call("SET", full_path, entry, "EX", ttlSec)
+else
+    redis.call("SET", full_path, entry)
+end
+
+if not isSuperLargeDirectory and name ~= "" then
+    redis.call("ZADD", dir_list_key, "NX", zscore, name)
+end
+
+return 0

weed/filer/redis_lua/universal_redis_store.go

@@ -0,0 +1,191 @@
+package redis_lua
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/go-redis/redis/v8"
+
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"github.com/chrislusf/seaweedfs/weed/filer/redis_lua/stored_procedure"
+	"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 UniversalRedisLuaStore struct {
+	Client                  redis.UniversalClient
+	superLargeDirectoryHash map[string]bool
+}
+
+func (store *UniversalRedisLuaStore) isSuperLargeDirectory(dir string) (isSuperLargeDirectory bool) {
+	_, isSuperLargeDirectory = store.superLargeDirectoryHash[dir]
+	return
+}
+
+func (store *UniversalRedisLuaStore) loadSuperLargeDirectories(superLargeDirectories []string) {
+	// set directory hash
+	store.superLargeDirectoryHash = make(map[string]bool)
+	for _, dir := range superLargeDirectories {
+		store.superLargeDirectoryHash[dir] = true
+	}
+}
+
+func (store *UniversalRedisLuaStore) BeginTransaction(ctx context.Context) (context.Context, error) {
+	return ctx, nil
+}
+func (store *UniversalRedisLuaStore) CommitTransaction(ctx context.Context) error {
+	return nil
+}
+func (store *UniversalRedisLuaStore) RollbackTransaction(ctx context.Context) error {
+	return nil
+}
+
+func (store *UniversalRedisLuaStore) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) {
+
+	value, err := entry.EncodeAttributesAndChunks()
+	if err != nil {
+		return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err)
+	}
+
+	if len(entry.Chunks) > 50 {
+		value = util.MaybeGzipData(value)
+	}
+
+	dir, name := entry.FullPath.DirAndName()
+
+	err = stored_procedure.InsertEntryScript.Run(ctx, store.Client,
+		[]string{string(entry.FullPath), genDirectoryListKey(dir)},
+		value, entry.TtlSec,
+		store.isSuperLargeDirectory(dir), 0, name).Err()
+
+	if err != nil {
+		return fmt.Errorf("persisting %s : %v", entry.FullPath, err)
+	}
+
+	return nil
+}
+
+func (store *UniversalRedisLuaStore) UpdateEntry(ctx context.Context, entry *filer.Entry) (err error) {
+
+	return store.InsertEntry(ctx, entry)
+}
+
+func (store *UniversalRedisLuaStore) FindEntry(ctx context.Context, fullpath util.FullPath) (entry *filer.Entry, err error) {
+
+	data, err := store.Client.Get(ctx, 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 = &filer.Entry{
+		FullPath: fullpath,
+	}
+	err = entry.DecodeAttributesAndChunks(util.MaybeDecompressData([]byte(data)))
+	if err != nil {
+		return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err)
+	}
+
+	return entry, nil
+}
+
+func (store *UniversalRedisLuaStore) DeleteEntry(ctx context.Context, fullpath util.FullPath) (err error) {
+
+	dir, name := fullpath.DirAndName()
+
+	err = stored_procedure.DeleteEntryScript.Run(ctx, store.Client,
+		[]string{string(fullpath), genDirectoryListKey(string(fullpath)), genDirectoryListKey(dir)},
+		store.isSuperLargeDirectory(dir), name).Err()
+
+	if err != nil {
+		return fmt.Errorf("DeleteEntry %s : %v", fullpath, err)
+	}
+
+	return nil
+}
+
+func (store *UniversalRedisLuaStore) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) (err error) {
+
+	if store.isSuperLargeDirectory(string(fullpath)) {
+		return nil
+	}
+
+	err = stored_procedure.DeleteFolderChildrenScript.Run(ctx, store.Client,
+		[]string{string(fullpath)}).Err()
+
+	if err != nil {
+		return fmt.Errorf("DeleteFolderChildren %s : %v", fullpath, err)
+	}
+
+	return nil
+}
+
+func (store *UniversalRedisLuaStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
+	return lastFileName, filer.ErrUnsupportedListDirectoryPrefixed
+}
+
+func (store *UniversalRedisLuaStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
+
+	dirListKey := genDirectoryListKey(string(dirPath))
+
+	min := "-"
+	if startFileName != "" {
+		if includeStartFile {
+			min = "[" + startFileName
+		} else {
+			min = "(" + startFileName
+		}
+	}
+
+	members, err := store.Client.ZRangeByLex(ctx, dirListKey, &redis.ZRangeBy{
+		Min:    min,
+		Max:    "+",
+		Offset: 0,
+		Count:  limit,
+	}).Result()
+	if err != nil {
+		return lastFileName, fmt.Errorf("list %s : %v", dirPath, err)
+	}
+
+	// fetch entry meta
+	for _, fileName := range members {
+		path := util.NewFullPath(string(dirPath), fileName)
+		entry, err := store.FindEntry(ctx, path)
+		lastFileName = fileName
+		if err != nil {
+			glog.V(0).Infof("list %s : %v", path, err)
+			if err == filer_pb.ErrNotFound {
+				continue
+			}
+		} else {
+			if entry.TtlSec > 0 {
+				if entry.Attr.Crtime.Add(time.Duration(entry.TtlSec) * time.Second).Before(time.Now()) {
+					store.DeleteEntry(ctx, path)
+					continue
+				}
+			}
+			if !eachEntryFunc(entry) {
+				break
+			}
+		}
+	}
+
+	return lastFileName, err
+}
+
+func genDirectoryListKey(dir string) (dirList string) {
+	return dir + DIR_LIST_MARKER
+}
+
+func (store *UniversalRedisLuaStore) Shutdown() {
+	store.Client.Close()
+}

weed/filer/redis_lua/universal_redis_store_kv.go

@@ -0,0 +1,42 @@
+package redis_lua
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"github.com/go-redis/redis/v8"
+)
+
+func (store *UniversalRedisLuaStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) {
+
+	_, err = store.Client.Set(ctx, string(key), value, 0).Result()
+
+	if err != nil {
+		return fmt.Errorf("kv put: %v", err)
+	}
+
+	return nil
+}
+
+func (store *UniversalRedisLuaStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) {
+
+	data, err := store.Client.Get(ctx, string(key)).Result()
+
+	if err == redis.Nil {
+		return nil, filer.ErrKvNotFound
+	}
+
+	return []byte(data), err
+}
+
+func (store *UniversalRedisLuaStore) KvDelete(ctx context.Context, key []byte) (err error) {
+
+	_, err = store.Client.Del(ctx, string(key)).Result()
+
+	if err != nil {
+		return fmt.Errorf("kv delete: %v", err)
+	}
+
+	return nil
+}