universal_redis_store.go 5.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186
  1. package redis3
  2. import (
  3. "context"
  4. "fmt"
  5. "time"
  6. "github.com/redis/go-redis/v9"
  7. redsync "github.com/go-redsync/redsync/v4"
  8. "github.com/seaweedfs/seaweedfs/weed/filer"
  9. "github.com/seaweedfs/seaweedfs/weed/glog"
  10. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  11. "github.com/seaweedfs/seaweedfs/weed/util"
  12. )
  13. const (
  14. DIR_LIST_MARKER = "\x00"
  15. )
  16. type UniversalRedis3Store struct {
  17. Client redis.UniversalClient
  18. redsync *redsync.Redsync
  19. }
  20. func (store *UniversalRedis3Store) BeginTransaction(ctx context.Context) (context.Context, error) {
  21. return ctx, nil
  22. }
  23. func (store *UniversalRedis3Store) CommitTransaction(ctx context.Context) error {
  24. return nil
  25. }
  26. func (store *UniversalRedis3Store) RollbackTransaction(ctx context.Context) error {
  27. return nil
  28. }
  29. func (store *UniversalRedis3Store) 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. if err = insertChild(ctx, store, genDirectoryListKey(dir), name); err != nil {
  36. return fmt.Errorf("persisting %s in parent dir: %v", entry.FullPath, err)
  37. }
  38. }
  39. return nil
  40. }
  41. func (store *UniversalRedis3Store) doInsertEntry(ctx context.Context, entry *filer.Entry) error {
  42. value, err := entry.EncodeAttributesAndChunks()
  43. if err != nil {
  44. return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err)
  45. }
  46. if len(entry.GetChunks()) > filer.CountEntryChunksForGzip {
  47. value = util.MaybeGzipData(value)
  48. }
  49. if err = store.Client.Set(ctx, string(entry.FullPath), value, time.Duration(entry.TtlSec)*time.Second).Err(); err != nil {
  50. return fmt.Errorf("persisting %s : %v", entry.FullPath, err)
  51. }
  52. return nil
  53. }
  54. func (store *UniversalRedis3Store) UpdateEntry(ctx context.Context, entry *filer.Entry) (err error) {
  55. return store.doInsertEntry(ctx, entry)
  56. }
  57. func (store *UniversalRedis3Store) FindEntry(ctx context.Context, fullpath util.FullPath) (entry *filer.Entry, err error) {
  58. data, err := store.Client.Get(ctx, string(fullpath)).Result()
  59. if err == redis.Nil {
  60. return nil, filer_pb.ErrNotFound
  61. }
  62. if err != nil {
  63. return nil, fmt.Errorf("get %s : %v", fullpath, err)
  64. }
  65. entry = &filer.Entry{
  66. FullPath: fullpath,
  67. }
  68. err = entry.DecodeAttributesAndChunks(util.MaybeDecompressData([]byte(data)))
  69. if err != nil {
  70. return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err)
  71. }
  72. return entry, nil
  73. }
  74. func (store *UniversalRedis3Store) DeleteEntry(ctx context.Context, fullpath util.FullPath) (err error) {
  75. _, err = store.Client.Del(ctx, genDirectoryListKey(string(fullpath))).Result()
  76. if err != nil {
  77. return fmt.Errorf("delete dir list %s : %v", fullpath, err)
  78. }
  79. _, err = store.Client.Del(ctx, string(fullpath)).Result()
  80. if err != nil {
  81. return fmt.Errorf("delete %s : %v", fullpath, err)
  82. }
  83. dir, name := fullpath.DirAndName()
  84. if name != "" {
  85. if err = removeChild(ctx, store, genDirectoryListKey(dir), name); err != nil {
  86. return fmt.Errorf("DeleteEntry %s in parent dir: %v", fullpath, err)
  87. }
  88. }
  89. return nil
  90. }
  91. func (store *UniversalRedis3Store) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) (err error) {
  92. return removeChildren(ctx, store, genDirectoryListKey(string(fullpath)), func(name string) error {
  93. path := util.NewFullPath(string(fullpath), name)
  94. _, err = store.Client.Del(ctx, string(path)).Result()
  95. if err != nil {
  96. return fmt.Errorf("DeleteFolderChildren %s in parent dir: %v", fullpath, err)
  97. }
  98. // not efficient, but need to remove if it is a directory
  99. store.Client.Del(ctx, genDirectoryListKey(string(path)))
  100. return nil
  101. })
  102. }
  103. func (store *UniversalRedis3Store) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
  104. return lastFileName, filer.ErrUnsupportedListDirectoryPrefixed
  105. }
  106. func (store *UniversalRedis3Store) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
  107. dirListKey := genDirectoryListKey(string(dirPath))
  108. counter := int64(0)
  109. err = listChildren(ctx, store, dirListKey, startFileName, func(fileName string) bool {
  110. if startFileName != "" {
  111. if !includeStartFile && startFileName == fileName {
  112. return true
  113. }
  114. }
  115. path := util.NewFullPath(string(dirPath), fileName)
  116. entry, err := store.FindEntry(ctx, path)
  117. lastFileName = fileName
  118. if err != nil {
  119. glog.V(0).Infof("list %s : %v", path, err)
  120. if err == filer_pb.ErrNotFound {
  121. return true
  122. }
  123. } else {
  124. if entry.TtlSec > 0 {
  125. if entry.Attr.Crtime.Add(time.Duration(entry.TtlSec) * time.Second).Before(time.Now()) {
  126. store.Client.Del(ctx, string(path)).Result()
  127. store.Client.ZRem(ctx, dirListKey, fileName).Result()
  128. return true
  129. }
  130. }
  131. counter++
  132. if !eachEntryFunc(entry) {
  133. return false
  134. }
  135. if counter >= limit {
  136. return false
  137. }
  138. }
  139. return true
  140. })
  141. return lastFileName, err
  142. }
  143. func genDirectoryListKey(dir string) (dirList string) {
  144. return dir + DIR_LIST_MARKER
  145. }
  146. func (store *UniversalRedis3Store) Shutdown() {
  147. store.Client.Close()
  148. }