You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

164 lines
5.8 KiB

7 years ago
7 years ago
5 years ago
7 years ago
6 years ago
6 years ago
6 years ago
6 years ago
5 years ago
6 years ago
6 years ago
6 years ago
5 years ago
6 years ago
6 years ago
5 years ago
6 years ago
6 years ago
  1. package filer2
  2. import (
  3. "context"
  4. "time"
  5. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  6. "github.com/chrislusf/seaweedfs/weed/stats"
  7. "github.com/chrislusf/seaweedfs/weed/util"
  8. )
  9. type FilerStore interface {
  10. // GetName gets the name to locate the configuration in filer.toml file
  11. GetName() string
  12. // Initialize initializes the file store
  13. Initialize(configuration util.Configuration, prefix string) error
  14. InsertEntry(context.Context, *Entry) error
  15. UpdateEntry(context.Context, *Entry) (err error)
  16. // err == filer2.ErrNotFound if not found
  17. FindEntry(context.Context, util.FullPath) (entry *Entry, err error)
  18. DeleteEntry(context.Context, util.FullPath) (err error)
  19. DeleteFolderChildren(context.Context, util.FullPath) (err error)
  20. ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int) ([]*Entry, error)
  21. ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int, prefix string) ([]*Entry, error)
  22. BeginTransaction(ctx context.Context) (context.Context, error)
  23. CommitTransaction(ctx context.Context) error
  24. RollbackTransaction(ctx context.Context) error
  25. Shutdown()
  26. }
  27. type FilerLocalStore interface {
  28. UpdateOffset(filer string, lastTsNs int64) error
  29. ReadOffset(filer string) (lastTsNs int64, err error)
  30. }
  31. type FilerStoreWrapper struct {
  32. ActualStore FilerStore
  33. }
  34. func NewFilerStoreWrapper(store FilerStore) *FilerStoreWrapper {
  35. if innerStore, ok := store.(*FilerStoreWrapper); ok {
  36. return innerStore
  37. }
  38. return &FilerStoreWrapper{
  39. ActualStore: store,
  40. }
  41. }
  42. func (fsw *FilerStoreWrapper) GetName() string {
  43. return fsw.ActualStore.GetName()
  44. }
  45. func (fsw *FilerStoreWrapper) Initialize(configuration util.Configuration, prefix string) error {
  46. return fsw.ActualStore.Initialize(configuration, prefix)
  47. }
  48. func (fsw *FilerStoreWrapper) InsertEntry(ctx context.Context, entry *Entry) error {
  49. stats.FilerStoreCounter.WithLabelValues(fsw.ActualStore.GetName(), "insert").Inc()
  50. start := time.Now()
  51. defer func() {
  52. stats.FilerStoreHistogram.WithLabelValues(fsw.ActualStore.GetName(), "insert").Observe(time.Since(start).Seconds())
  53. }()
  54. filer_pb.BeforeEntrySerialization(entry.Chunks)
  55. return fsw.ActualStore.InsertEntry(ctx, entry)
  56. }
  57. func (fsw *FilerStoreWrapper) UpdateEntry(ctx context.Context, entry *Entry) error {
  58. stats.FilerStoreCounter.WithLabelValues(fsw.ActualStore.GetName(), "update").Inc()
  59. start := time.Now()
  60. defer func() {
  61. stats.FilerStoreHistogram.WithLabelValues(fsw.ActualStore.GetName(), "update").Observe(time.Since(start).Seconds())
  62. }()
  63. filer_pb.BeforeEntrySerialization(entry.Chunks)
  64. return fsw.ActualStore.UpdateEntry(ctx, entry)
  65. }
  66. func (fsw *FilerStoreWrapper) FindEntry(ctx context.Context, fp util.FullPath) (entry *Entry, err error) {
  67. stats.FilerStoreCounter.WithLabelValues(fsw.ActualStore.GetName(), "find").Inc()
  68. start := time.Now()
  69. defer func() {
  70. stats.FilerStoreHistogram.WithLabelValues(fsw.ActualStore.GetName(), "find").Observe(time.Since(start).Seconds())
  71. }()
  72. entry, err = fsw.ActualStore.FindEntry(ctx, fp)
  73. if err != nil {
  74. return nil, err
  75. }
  76. filer_pb.AfterEntryDeserialization(entry.Chunks)
  77. return
  78. }
  79. func (fsw *FilerStoreWrapper) DeleteEntry(ctx context.Context, fp util.FullPath) (err error) {
  80. stats.FilerStoreCounter.WithLabelValues(fsw.ActualStore.GetName(), "delete").Inc()
  81. start := time.Now()
  82. defer func() {
  83. stats.FilerStoreHistogram.WithLabelValues(fsw.ActualStore.GetName(), "delete").Observe(time.Since(start).Seconds())
  84. }()
  85. return fsw.ActualStore.DeleteEntry(ctx, fp)
  86. }
  87. func (fsw *FilerStoreWrapper) DeleteFolderChildren(ctx context.Context, fp util.FullPath) (err error) {
  88. stats.FilerStoreCounter.WithLabelValues(fsw.ActualStore.GetName(), "deleteFolderChildren").Inc()
  89. start := time.Now()
  90. defer func() {
  91. stats.FilerStoreHistogram.WithLabelValues(fsw.ActualStore.GetName(), "deleteFolderChildren").Observe(time.Since(start).Seconds())
  92. }()
  93. return fsw.ActualStore.DeleteFolderChildren(ctx, fp)
  94. }
  95. func (fsw *FilerStoreWrapper) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int) ([]*Entry, error) {
  96. stats.FilerStoreCounter.WithLabelValues(fsw.ActualStore.GetName(), "list").Inc()
  97. start := time.Now()
  98. defer func() {
  99. stats.FilerStoreHistogram.WithLabelValues(fsw.ActualStore.GetName(), "list").Observe(time.Since(start).Seconds())
  100. }()
  101. entries, err := fsw.ActualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit)
  102. if err != nil {
  103. return nil, err
  104. }
  105. for _, entry := range entries {
  106. filer_pb.AfterEntryDeserialization(entry.Chunks)
  107. }
  108. return entries, err
  109. }
  110. func (fsw *FilerStoreWrapper) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int, prefix string) ([]*Entry, error) {
  111. stats.FilerStoreCounter.WithLabelValues(fsw.ActualStore.GetName(), "list").Inc()
  112. start := time.Now()
  113. defer func() {
  114. stats.FilerStoreHistogram.WithLabelValues(fsw.ActualStore.GetName(), "list").Observe(time.Since(start).Seconds())
  115. }()
  116. entries, err := fsw.ActualStore.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, prefix)
  117. if err != nil {
  118. return nil, err
  119. }
  120. for _, entry := range entries {
  121. filer_pb.AfterEntryDeserialization(entry.Chunks)
  122. }
  123. return entries, nil
  124. }
  125. func (fsw *FilerStoreWrapper) BeginTransaction(ctx context.Context) (context.Context, error) {
  126. return fsw.ActualStore.BeginTransaction(ctx)
  127. }
  128. func (fsw *FilerStoreWrapper) CommitTransaction(ctx context.Context) error {
  129. return fsw.ActualStore.CommitTransaction(ctx)
  130. }
  131. func (fsw *FilerStoreWrapper) RollbackTransaction(ctx context.Context) error {
  132. return fsw.ActualStore.RollbackTransaction(ctx)
  133. }
  134. func (fsw *FilerStoreWrapper) Shutdown() {
  135. fsw.ActualStore.Shutdown()
  136. }