universal_redis_store.go 6.1 KB

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