filerstore_wrapper.go 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349
  1. package filer
  2. import (
  3. "context"
  4. "github.com/seaweedfs/seaweedfs/weed/glog"
  5. "github.com/viant/ptrie"
  6. "io"
  7. "math"
  8. "strings"
  9. "time"
  10. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  11. "github.com/seaweedfs/seaweedfs/weed/stats"
  12. "github.com/seaweedfs/seaweedfs/weed/util"
  13. )
  14. var (
  15. _ = VirtualFilerStore(&FilerStoreWrapper{})
  16. _ = Debuggable(&FilerStoreWrapper{})
  17. )
  18. type VirtualFilerStore interface {
  19. FilerStore
  20. DeleteHardLink(ctx context.Context, hardLinkId HardLinkId) error
  21. DeleteOneEntry(ctx context.Context, entry *Entry) error
  22. AddPathSpecificStore(path string, storeId string, store FilerStore)
  23. OnBucketCreation(bucket string)
  24. OnBucketDeletion(bucket string)
  25. CanDropWholeBucket() bool
  26. }
  27. type FilerStoreWrapper struct {
  28. defaultStore FilerStore
  29. pathToStore ptrie.Trie
  30. storeIdToStore map[string]FilerStore
  31. }
  32. func NewFilerStoreWrapper(store FilerStore) *FilerStoreWrapper {
  33. if innerStore, ok := store.(*FilerStoreWrapper); ok {
  34. return innerStore
  35. }
  36. return &FilerStoreWrapper{
  37. defaultStore: store,
  38. pathToStore: ptrie.New(),
  39. storeIdToStore: make(map[string]FilerStore),
  40. }
  41. }
  42. func (fsw *FilerStoreWrapper) CanDropWholeBucket() bool {
  43. if ba, ok := fsw.defaultStore.(BucketAware); ok {
  44. return ba.CanDropWholeBucket()
  45. }
  46. return false
  47. }
  48. func (fsw *FilerStoreWrapper) OnBucketCreation(bucket string) {
  49. for _, store := range fsw.storeIdToStore {
  50. if ba, ok := store.(BucketAware); ok {
  51. ba.OnBucketCreation(bucket)
  52. }
  53. }
  54. if ba, ok := fsw.defaultStore.(BucketAware); ok {
  55. ba.OnBucketCreation(bucket)
  56. }
  57. }
  58. func (fsw *FilerStoreWrapper) OnBucketDeletion(bucket string) {
  59. for _, store := range fsw.storeIdToStore {
  60. if ba, ok := store.(BucketAware); ok {
  61. ba.OnBucketDeletion(bucket)
  62. }
  63. }
  64. if ba, ok := fsw.defaultStore.(BucketAware); ok {
  65. ba.OnBucketDeletion(bucket)
  66. }
  67. }
  68. func (fsw *FilerStoreWrapper) AddPathSpecificStore(path string, storeId string, store FilerStore) {
  69. fsw.storeIdToStore[storeId] = NewFilerStorePathTranslator(path, store)
  70. err := fsw.pathToStore.Put([]byte(path), storeId)
  71. if err != nil {
  72. glog.Fatalf("put path specific store: %v", err)
  73. }
  74. }
  75. func (fsw *FilerStoreWrapper) getActualStore(path util.FullPath) (store FilerStore) {
  76. store = fsw.defaultStore
  77. if path == "/" {
  78. return
  79. }
  80. var storeId string
  81. fsw.pathToStore.MatchPrefix([]byte(path), func(key []byte, value interface{}) bool {
  82. storeId = value.(string)
  83. return false
  84. })
  85. if storeId != "" {
  86. store = fsw.storeIdToStore[storeId]
  87. }
  88. return
  89. }
  90. func (fsw *FilerStoreWrapper) getDefaultStore() (store FilerStore) {
  91. return fsw.defaultStore
  92. }
  93. func (fsw *FilerStoreWrapper) GetName() string {
  94. return fsw.getDefaultStore().GetName()
  95. }
  96. func (fsw *FilerStoreWrapper) Initialize(configuration util.Configuration, prefix string) error {
  97. return fsw.getDefaultStore().Initialize(configuration, prefix)
  98. }
  99. func (fsw *FilerStoreWrapper) InsertEntry(ctx context.Context, entry *Entry) error {
  100. actualStore := fsw.getActualStore(entry.FullPath)
  101. stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "insert").Inc()
  102. start := time.Now()
  103. defer func() {
  104. stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "insert").Observe(time.Since(start).Seconds())
  105. }()
  106. filer_pb.BeforeEntrySerialization(entry.GetChunks())
  107. if entry.Mime == "application/octet-stream" {
  108. entry.Mime = ""
  109. }
  110. if err := fsw.handleUpdateToHardLinks(ctx, entry); err != nil {
  111. return err
  112. }
  113. // glog.V(4).Infof("InsertEntry %s", entry.FullPath)
  114. return actualStore.InsertEntry(ctx, entry)
  115. }
  116. func (fsw *FilerStoreWrapper) UpdateEntry(ctx context.Context, entry *Entry) error {
  117. actualStore := fsw.getActualStore(entry.FullPath)
  118. stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "update").Inc()
  119. start := time.Now()
  120. defer func() {
  121. stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "update").Observe(time.Since(start).Seconds())
  122. }()
  123. filer_pb.BeforeEntrySerialization(entry.GetChunks())
  124. if entry.Mime == "application/octet-stream" {
  125. entry.Mime = ""
  126. }
  127. if err := fsw.handleUpdateToHardLinks(ctx, entry); err != nil {
  128. return err
  129. }
  130. // glog.V(4).Infof("UpdateEntry %s", entry.FullPath)
  131. return actualStore.UpdateEntry(ctx, entry)
  132. }
  133. func (fsw *FilerStoreWrapper) FindEntry(ctx context.Context, fp util.FullPath) (entry *Entry, err error) {
  134. actualStore := fsw.getActualStore(fp)
  135. stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "find").Inc()
  136. start := time.Now()
  137. defer func() {
  138. stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "find").Observe(time.Since(start).Seconds())
  139. }()
  140. entry, err = actualStore.FindEntry(ctx, fp)
  141. // glog.V(4).Infof("FindEntry %s: %v", fp, err)
  142. if err != nil {
  143. return nil, err
  144. }
  145. fsw.maybeReadHardLink(ctx, entry)
  146. filer_pb.AfterEntryDeserialization(entry.GetChunks())
  147. return
  148. }
  149. func (fsw *FilerStoreWrapper) DeleteEntry(ctx context.Context, fp util.FullPath) (err error) {
  150. actualStore := fsw.getActualStore(fp)
  151. stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "delete").Inc()
  152. start := time.Now()
  153. defer func() {
  154. stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "delete").Observe(time.Since(start).Seconds())
  155. }()
  156. existingEntry, findErr := fsw.FindEntry(ctx, fp)
  157. if findErr == filer_pb.ErrNotFound {
  158. return nil
  159. }
  160. if len(existingEntry.HardLinkId) != 0 {
  161. // remove hard link
  162. glog.V(4).Infof("DeleteHardLink %s", existingEntry.FullPath)
  163. if err = fsw.DeleteHardLink(ctx, existingEntry.HardLinkId); err != nil {
  164. return err
  165. }
  166. }
  167. // glog.V(4).Infof("DeleteEntry %s", fp)
  168. return actualStore.DeleteEntry(ctx, fp)
  169. }
  170. func (fsw *FilerStoreWrapper) DeleteOneEntry(ctx context.Context, existingEntry *Entry) (err error) {
  171. actualStore := fsw.getActualStore(existingEntry.FullPath)
  172. stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "delete").Inc()
  173. start := time.Now()
  174. defer func() {
  175. stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "delete").Observe(time.Since(start).Seconds())
  176. }()
  177. if len(existingEntry.HardLinkId) != 0 {
  178. // remove hard link
  179. glog.V(4).Infof("DeleteHardLink %s", existingEntry.FullPath)
  180. if err = fsw.DeleteHardLink(ctx, existingEntry.HardLinkId); err != nil {
  181. return err
  182. }
  183. }
  184. // glog.V(4).Infof("DeleteOneEntry %s", existingEntry.FullPath)
  185. return actualStore.DeleteEntry(ctx, existingEntry.FullPath)
  186. }
  187. func (fsw *FilerStoreWrapper) DeleteFolderChildren(ctx context.Context, fp util.FullPath) (err error) {
  188. actualStore := fsw.getActualStore(fp + "/")
  189. stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "deleteFolderChildren").Inc()
  190. start := time.Now()
  191. defer func() {
  192. stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "deleteFolderChildren").Observe(time.Since(start).Seconds())
  193. }()
  194. // glog.V(4).Infof("DeleteFolderChildren %s", fp)
  195. return actualStore.DeleteFolderChildren(ctx, fp)
  196. }
  197. func (fsw *FilerStoreWrapper) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc ListEachEntryFunc) (string, error) {
  198. actualStore := fsw.getActualStore(dirPath + "/")
  199. stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "list").Inc()
  200. start := time.Now()
  201. defer func() {
  202. stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "list").Observe(time.Since(start).Seconds())
  203. }()
  204. // glog.V(4).Infof("ListDirectoryEntries %s from %s limit %d", dirPath, startFileName, limit)
  205. return actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *Entry) bool {
  206. fsw.maybeReadHardLink(ctx, entry)
  207. filer_pb.AfterEntryDeserialization(entry.GetChunks())
  208. return eachEntryFunc(entry)
  209. })
  210. }
  211. func (fsw *FilerStoreWrapper) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc ListEachEntryFunc) (lastFileName string, err error) {
  212. actualStore := fsw.getActualStore(dirPath + "/")
  213. stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "prefixList").Inc()
  214. start := time.Now()
  215. defer func() {
  216. stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "prefixList").Observe(time.Since(start).Seconds())
  217. }()
  218. if limit > math.MaxInt32-1 {
  219. limit = math.MaxInt32 - 1
  220. }
  221. // glog.V(4).Infof("ListDirectoryPrefixedEntries %s from %s prefix %s limit %d", dirPath, startFileName, prefix, limit)
  222. adjustedEntryFunc := func(entry *Entry) bool {
  223. fsw.maybeReadHardLink(ctx, entry)
  224. filer_pb.AfterEntryDeserialization(entry.GetChunks())
  225. return eachEntryFunc(entry)
  226. }
  227. lastFileName, err = actualStore.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, prefix, adjustedEntryFunc)
  228. if err == ErrUnsupportedListDirectoryPrefixed {
  229. lastFileName, err = fsw.prefixFilterEntries(ctx, dirPath, startFileName, includeStartFile, limit, prefix, adjustedEntryFunc)
  230. }
  231. return lastFileName, err
  232. }
  233. func (fsw *FilerStoreWrapper) prefixFilterEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc ListEachEntryFunc) (lastFileName string, err error) {
  234. actualStore := fsw.getActualStore(dirPath + "/")
  235. if prefix == "" {
  236. return actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, eachEntryFunc)
  237. }
  238. var notPrefixed []*Entry
  239. lastFileName, err = actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *Entry) bool {
  240. notPrefixed = append(notPrefixed, entry)
  241. return true
  242. })
  243. if err != nil {
  244. return
  245. }
  246. count := int64(0)
  247. for count < limit && len(notPrefixed) > 0 {
  248. var isLastItemHasPrefix bool
  249. for _, entry := range notPrefixed {
  250. if strings.HasPrefix(entry.Name(), prefix) {
  251. isLastItemHasPrefix = true
  252. count++
  253. if !eachEntryFunc(entry) {
  254. return
  255. }
  256. if count >= limit {
  257. break
  258. }
  259. } else {
  260. isLastItemHasPrefix = false
  261. }
  262. }
  263. if count < limit && isLastItemHasPrefix && len(notPrefixed) == int(limit) {
  264. notPrefixed = notPrefixed[:0]
  265. lastFileName, err = actualStore.ListDirectoryEntries(ctx, dirPath, lastFileName, false, limit, func(entry *Entry) bool {
  266. notPrefixed = append(notPrefixed, entry)
  267. return true
  268. })
  269. if err != nil {
  270. return
  271. }
  272. } else {
  273. break
  274. }
  275. }
  276. return
  277. }
  278. func (fsw *FilerStoreWrapper) BeginTransaction(ctx context.Context) (context.Context, error) {
  279. return fsw.getDefaultStore().BeginTransaction(ctx)
  280. }
  281. func (fsw *FilerStoreWrapper) CommitTransaction(ctx context.Context) error {
  282. return fsw.getDefaultStore().CommitTransaction(ctx)
  283. }
  284. func (fsw *FilerStoreWrapper) RollbackTransaction(ctx context.Context) error {
  285. return fsw.getDefaultStore().RollbackTransaction(ctx)
  286. }
  287. func (fsw *FilerStoreWrapper) Shutdown() {
  288. fsw.getDefaultStore().Shutdown()
  289. }
  290. func (fsw *FilerStoreWrapper) KvPut(ctx context.Context, key []byte, value []byte) (err error) {
  291. return fsw.getDefaultStore().KvPut(ctx, key, value)
  292. }
  293. func (fsw *FilerStoreWrapper) KvGet(ctx context.Context, key []byte) (value []byte, err error) {
  294. return fsw.getDefaultStore().KvGet(ctx, key)
  295. }
  296. func (fsw *FilerStoreWrapper) KvDelete(ctx context.Context, key []byte) (err error) {
  297. return fsw.getDefaultStore().KvDelete(ctx, key)
  298. }
  299. func (fsw *FilerStoreWrapper) Debug(writer io.Writer) {
  300. if debuggable, ok := fsw.getDefaultStore().(Debuggable); ok {
  301. debuggable.Debug(writer)
  302. }
  303. }