filerstore_wrapper.go 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352
  1. package filer
  2. import (
  3. "context"
  4. "io"
  5. "math"
  6. "strings"
  7. "time"
  8. "github.com/seaweedfs/seaweedfs/weed/glog"
  9. "github.com/viant/ptrie"
  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[string]
  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[string](),
  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 == "/" || path == "//" {
  78. return
  79. }
  80. var storeId string
  81. fsw.pathToStore.MatchPrefix([]byte(path), func(key []byte, value string) bool {
  82. storeId = value
  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 || existingEntry == nil {
  158. return nil
  159. }
  160. if len(existingEntry.HardLinkId) != 0 {
  161. // remove hard link
  162. op := ctx.Value("OP")
  163. if op != "MV" {
  164. glog.V(4).Infof("DeleteHardLink %s", existingEntry.FullPath)
  165. if err = fsw.DeleteHardLink(ctx, existingEntry.HardLinkId); err != nil {
  166. return err
  167. }
  168. }
  169. }
  170. // glog.V(4).Infof("DeleteEntry %s", fp)
  171. return actualStore.DeleteEntry(ctx, fp)
  172. }
  173. func (fsw *FilerStoreWrapper) DeleteOneEntry(ctx context.Context, existingEntry *Entry) (err error) {
  174. actualStore := fsw.getActualStore(existingEntry.FullPath)
  175. stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "delete").Inc()
  176. start := time.Now()
  177. defer func() {
  178. stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "delete").Observe(time.Since(start).Seconds())
  179. }()
  180. if len(existingEntry.HardLinkId) != 0 {
  181. // remove hard link
  182. op := ctx.Value("OP")
  183. if op != "MV" {
  184. glog.V(4).Infof("DeleteHardLink %s", existingEntry.FullPath)
  185. if err = fsw.DeleteHardLink(ctx, existingEntry.HardLinkId); err != nil {
  186. return err
  187. }
  188. }
  189. }
  190. // glog.V(4).Infof("DeleteOneEntry %s", existingEntry.FullPath)
  191. return actualStore.DeleteEntry(ctx, existingEntry.FullPath)
  192. }
  193. func (fsw *FilerStoreWrapper) DeleteFolderChildren(ctx context.Context, fp util.FullPath) (err error) {
  194. actualStore := fsw.getActualStore(fp + "/")
  195. stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "deleteFolderChildren").Inc()
  196. start := time.Now()
  197. defer func() {
  198. stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "deleteFolderChildren").Observe(time.Since(start).Seconds())
  199. }()
  200. // glog.V(4).Infof("DeleteFolderChildren %s", fp)
  201. return actualStore.DeleteFolderChildren(ctx, fp)
  202. }
  203. func (fsw *FilerStoreWrapper) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc ListEachEntryFunc) (string, error) {
  204. actualStore := fsw.getActualStore(dirPath + "/")
  205. stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "list").Inc()
  206. start := time.Now()
  207. defer func() {
  208. stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "list").Observe(time.Since(start).Seconds())
  209. }()
  210. // glog.V(4).Infof("ListDirectoryEntries %s from %s limit %d", dirPath, startFileName, limit)
  211. return actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *Entry) bool {
  212. fsw.maybeReadHardLink(ctx, entry)
  213. filer_pb.AfterEntryDeserialization(entry.GetChunks())
  214. return eachEntryFunc(entry)
  215. })
  216. }
  217. func (fsw *FilerStoreWrapper) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc ListEachEntryFunc) (lastFileName string, err error) {
  218. actualStore := fsw.getActualStore(dirPath + "/")
  219. stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "prefixList").Inc()
  220. start := time.Now()
  221. defer func() {
  222. stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "prefixList").Observe(time.Since(start).Seconds())
  223. }()
  224. if limit > math.MaxInt32-1 {
  225. limit = math.MaxInt32 - 1
  226. }
  227. // glog.V(4).Infof("ListDirectoryPrefixedEntries %s from %s prefix %s limit %d", dirPath, startFileName, prefix, limit)
  228. adjustedEntryFunc := func(entry *Entry) bool {
  229. fsw.maybeReadHardLink(ctx, entry)
  230. filer_pb.AfterEntryDeserialization(entry.GetChunks())
  231. return eachEntryFunc(entry)
  232. }
  233. lastFileName, err = actualStore.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, prefix, adjustedEntryFunc)
  234. if err == ErrUnsupportedListDirectoryPrefixed {
  235. lastFileName, err = fsw.prefixFilterEntries(ctx, dirPath, startFileName, includeStartFile, limit, prefix, adjustedEntryFunc)
  236. }
  237. return lastFileName, err
  238. }
  239. func (fsw *FilerStoreWrapper) prefixFilterEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc ListEachEntryFunc) (lastFileName string, err error) {
  240. actualStore := fsw.getActualStore(dirPath + "/")
  241. if prefix == "" {
  242. return actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, eachEntryFunc)
  243. }
  244. var notPrefixed []*Entry
  245. lastFileName, err = actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *Entry) bool {
  246. notPrefixed = append(notPrefixed, entry)
  247. return true
  248. })
  249. if err != nil {
  250. return
  251. }
  252. count := int64(0)
  253. for count < limit && len(notPrefixed) > 0 {
  254. for _, entry := range notPrefixed {
  255. if strings.HasPrefix(entry.Name(), prefix) {
  256. count++
  257. if !eachEntryFunc(entry) {
  258. return
  259. }
  260. if count >= limit {
  261. break
  262. }
  263. }
  264. }
  265. if count < limit && lastFileName < prefix {
  266. notPrefixed = notPrefixed[:0]
  267. lastFileName, err = actualStore.ListDirectoryEntries(ctx, dirPath, lastFileName, false, limit, func(entry *Entry) bool {
  268. notPrefixed = append(notPrefixed, entry)
  269. return true
  270. })
  271. if err != nil {
  272. return
  273. }
  274. } else {
  275. break
  276. }
  277. }
  278. return
  279. }
  280. func (fsw *FilerStoreWrapper) BeginTransaction(ctx context.Context) (context.Context, error) {
  281. return fsw.getDefaultStore().BeginTransaction(ctx)
  282. }
  283. func (fsw *FilerStoreWrapper) CommitTransaction(ctx context.Context) error {
  284. return fsw.getDefaultStore().CommitTransaction(ctx)
  285. }
  286. func (fsw *FilerStoreWrapper) RollbackTransaction(ctx context.Context) error {
  287. return fsw.getDefaultStore().RollbackTransaction(ctx)
  288. }
  289. func (fsw *FilerStoreWrapper) Shutdown() {
  290. fsw.getDefaultStore().Shutdown()
  291. }
  292. func (fsw *FilerStoreWrapper) KvPut(ctx context.Context, key []byte, value []byte) (err error) {
  293. return fsw.getDefaultStore().KvPut(ctx, key, value)
  294. }
  295. func (fsw *FilerStoreWrapper) KvGet(ctx context.Context, key []byte) (value []byte, err error) {
  296. return fsw.getDefaultStore().KvGet(ctx, key)
  297. }
  298. func (fsw *FilerStoreWrapper) KvDelete(ctx context.Context, key []byte) (err error) {
  299. return fsw.getDefaultStore().KvDelete(ctx, key)
  300. }
  301. func (fsw *FilerStoreWrapper) Debug(writer io.Writer) {
  302. if debuggable, ok := fsw.getDefaultStore().(Debuggable); ok {
  303. debuggable.Debug(writer)
  304. }
  305. }