universal_redis_store.go 5.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209
  1. package redis
  2. import (
  3. "context"
  4. "fmt"
  5. "golang.org/x/exp/slices"
  6. "strings"
  7. "time"
  8. "github.com/redis/go-redis/v9"
  9. "github.com/seaweedfs/seaweedfs/weed/filer"
  10. "github.com/seaweedfs/seaweedfs/weed/glog"
  11. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  12. "github.com/seaweedfs/seaweedfs/weed/util"
  13. )
  14. const (
  15. DIR_LIST_MARKER = "\x00"
  16. )
  17. type UniversalRedisStore struct {
  18. Client redis.UniversalClient
  19. }
  20. func (store *UniversalRedisStore) BeginTransaction(ctx context.Context) (context.Context, error) {
  21. return ctx, nil
  22. }
  23. func (store *UniversalRedisStore) CommitTransaction(ctx context.Context) error {
  24. return nil
  25. }
  26. func (store *UniversalRedisStore) RollbackTransaction(ctx context.Context) error {
  27. return nil
  28. }
  29. func (store *UniversalRedisStore) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) {
  30. if err = store.doInsertEntry(ctx, entry); err != nil {
  31. return err
  32. }
  33. dir, name := entry.FullPath.DirAndName()
  34. if name != "" {
  35. _, err = store.Client.SAdd(ctx, genDirectoryListKey(dir), name).Result()
  36. if err != nil {
  37. return fmt.Errorf("persisting %s in parent dir: %v", entry.FullPath, err)
  38. }
  39. }
  40. return nil
  41. }
  42. func (store *UniversalRedisStore) doInsertEntry(ctx context.Context, entry *filer.Entry) error {
  43. value, err := entry.EncodeAttributesAndChunks()
  44. if err != nil {
  45. return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err)
  46. }
  47. if len(entry.GetChunks()) > filer.CountEntryChunksForGzip {
  48. value = util.MaybeGzipData(value)
  49. }
  50. _, err = store.Client.Set(ctx, string(entry.FullPath), value, time.Duration(entry.TtlSec)*time.Second).Result()
  51. if err != nil {
  52. return fmt.Errorf("persisting %s : %v", entry.FullPath, err)
  53. }
  54. return nil
  55. }
  56. func (store *UniversalRedisStore) UpdateEntry(ctx context.Context, entry *filer.Entry) (err error) {
  57. return store.doInsertEntry(ctx, entry)
  58. }
  59. func (store *UniversalRedisStore) FindEntry(ctx context.Context, fullpath util.FullPath) (entry *filer.Entry, err error) {
  60. data, err := store.Client.Get(ctx, string(fullpath)).Result()
  61. if err == redis.Nil {
  62. return nil, filer_pb.ErrNotFound
  63. }
  64. if err != nil {
  65. return nil, fmt.Errorf("get %s : %v", fullpath, err)
  66. }
  67. entry = &filer.Entry{
  68. FullPath: fullpath,
  69. }
  70. err = entry.DecodeAttributesAndChunks(util.MaybeDecompressData([]byte(data)))
  71. if err != nil {
  72. return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err)
  73. }
  74. return entry, nil
  75. }
  76. func (store *UniversalRedisStore) DeleteEntry(ctx context.Context, fullpath util.FullPath) (err error) {
  77. _, err = store.Client.Del(ctx, string(fullpath)).Result()
  78. if err != nil {
  79. return fmt.Errorf("delete %s : %v", fullpath, err)
  80. }
  81. dir, name := fullpath.DirAndName()
  82. if name != "" {
  83. _, err = store.Client.SRem(ctx, genDirectoryListKey(dir), name).Result()
  84. if err != nil {
  85. return fmt.Errorf("delete %s in parent dir: %v", fullpath, err)
  86. }
  87. }
  88. return nil
  89. }
  90. func (store *UniversalRedisStore) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) (err error) {
  91. members, err := store.Client.SMembers(ctx, genDirectoryListKey(string(fullpath))).Result()
  92. if err != nil {
  93. return fmt.Errorf("delete folder %s : %v", fullpath, err)
  94. }
  95. for _, fileName := range members {
  96. path := util.NewFullPath(string(fullpath), fileName)
  97. _, err = store.Client.Del(ctx, string(path)).Result()
  98. if err != nil {
  99. return fmt.Errorf("delete %s in parent dir: %v", fullpath, err)
  100. }
  101. // not efficient, but need to remove if it is a directory
  102. store.Client.Del(ctx, genDirectoryListKey(string(path)))
  103. }
  104. return nil
  105. }
  106. func (store *UniversalRedisStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
  107. return lastFileName, filer.ErrUnsupportedListDirectoryPrefixed
  108. }
  109. func (store *UniversalRedisStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
  110. dirListKey := genDirectoryListKey(string(dirPath))
  111. members, err := store.Client.SMembers(ctx, dirListKey).Result()
  112. if err != nil {
  113. return lastFileName, fmt.Errorf("list %s : %v", dirPath, err)
  114. }
  115. // skip
  116. if startFileName != "" {
  117. var t []string
  118. for _, m := range members {
  119. if strings.Compare(m, startFileName) >= 0 {
  120. if m == startFileName {
  121. if includeStartFile {
  122. t = append(t, m)
  123. }
  124. } else {
  125. t = append(t, m)
  126. }
  127. }
  128. }
  129. members = t
  130. }
  131. // sort
  132. slices.SortFunc(members, func(a, b string) int {
  133. return strings.Compare(a, b)
  134. })
  135. // limit
  136. if limit < int64(len(members)) {
  137. members = members[:limit]
  138. }
  139. // fetch entry meta
  140. for _, fileName := range members {
  141. path := util.NewFullPath(string(dirPath), fileName)
  142. entry, err := store.FindEntry(ctx, path)
  143. lastFileName = fileName
  144. if err != nil {
  145. glog.V(0).Infof("list %s : %v", path, err)
  146. if err == filer_pb.ErrNotFound {
  147. continue
  148. }
  149. } else {
  150. if entry.TtlSec > 0 {
  151. if entry.Attr.Crtime.Add(time.Duration(entry.TtlSec) * time.Second).Before(time.Now()) {
  152. store.Client.Del(ctx, string(path)).Result()
  153. store.Client.SRem(ctx, dirListKey, fileName).Result()
  154. continue
  155. }
  156. }
  157. if !eachEntryFunc(entry) {
  158. break
  159. }
  160. }
  161. }
  162. return lastFileName, err
  163. }
  164. func genDirectoryListKey(dir string) (dirList string) {
  165. return dir + DIR_LIST_MARKER
  166. }
  167. func (store *UniversalRedisStore) Shutdown() {
  168. store.Client.Close()
  169. }