From 708acee502918080b286cb96114a3789b9f1183d Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 15 Aug 2018 00:01:38 -0700 Subject: [PATCH] add support for redis cluster fix https://github.com/chrislusf/seaweedfs/issues/705 --- weed/filer2/configuration.go | 6 + weed/filer2/redis/redis_cluster_store.go | 31 +++++ weed/filer2/redis/redis_store.go | 133 +------------------ weed/filer2/redis/universal_redis_store.go | 141 +++++++++++++++++++++ 4 files changed, 179 insertions(+), 132 deletions(-) create mode 100644 weed/filer2/redis/redis_cluster_store.go create mode 100644 weed/filer2/redis/universal_redis_store.go 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 +}