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.
355 lines
11 KiB
355 lines
11 KiB
package filer
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
"math"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/glog"
|
|
"github.com/viant/ptrie"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/stats"
|
|
"github.com/seaweedfs/seaweedfs/weed/util"
|
|
)
|
|
|
|
var (
|
|
_ = VirtualFilerStore(&FilerStoreWrapper{})
|
|
_ = Debuggable(&FilerStoreWrapper{})
|
|
)
|
|
|
|
type VirtualFilerStore interface {
|
|
FilerStore
|
|
DeleteHardLink(ctx context.Context, hardLinkId HardLinkId) error
|
|
DeleteOneEntry(ctx context.Context, entry *Entry) error
|
|
AddPathSpecificStore(path string, storeId string, store FilerStore)
|
|
OnBucketCreation(bucket string)
|
|
OnBucketDeletion(bucket string)
|
|
CanDropWholeBucket() bool
|
|
}
|
|
|
|
type FilerStoreWrapper struct {
|
|
defaultStore FilerStore
|
|
pathToStore ptrie.Trie[string]
|
|
storeIdToStore map[string]FilerStore
|
|
}
|
|
|
|
func NewFilerStoreWrapper(store FilerStore) *FilerStoreWrapper {
|
|
if innerStore, ok := store.(*FilerStoreWrapper); ok {
|
|
return innerStore
|
|
}
|
|
return &FilerStoreWrapper{
|
|
defaultStore: store,
|
|
pathToStore: ptrie.New[string](),
|
|
storeIdToStore: make(map[string]FilerStore),
|
|
}
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) CanDropWholeBucket() bool {
|
|
if ba, ok := fsw.defaultStore.(BucketAware); ok {
|
|
return ba.CanDropWholeBucket()
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) OnBucketCreation(bucket string) {
|
|
for _, store := range fsw.storeIdToStore {
|
|
if ba, ok := store.(BucketAware); ok {
|
|
ba.OnBucketCreation(bucket)
|
|
}
|
|
}
|
|
if ba, ok := fsw.defaultStore.(BucketAware); ok {
|
|
ba.OnBucketCreation(bucket)
|
|
}
|
|
}
|
|
func (fsw *FilerStoreWrapper) OnBucketDeletion(bucket string) {
|
|
for _, store := range fsw.storeIdToStore {
|
|
if ba, ok := store.(BucketAware); ok {
|
|
ba.OnBucketDeletion(bucket)
|
|
}
|
|
}
|
|
if ba, ok := fsw.defaultStore.(BucketAware); ok {
|
|
ba.OnBucketDeletion(bucket)
|
|
}
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) AddPathSpecificStore(path string, storeId string, store FilerStore) {
|
|
fsw.storeIdToStore[storeId] = NewFilerStorePathTranslator(path, store)
|
|
err := fsw.pathToStore.Put([]byte(path), storeId)
|
|
if err != nil {
|
|
glog.Fatalf("put path specific store: %v", err)
|
|
}
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) getActualStore(path util.FullPath) (store FilerStore) {
|
|
store = fsw.defaultStore
|
|
if path == "/" || path == "//" {
|
|
return
|
|
}
|
|
var storeId string
|
|
fsw.pathToStore.MatchPrefix([]byte(path), func(key []byte, value string) bool {
|
|
storeId = value
|
|
return false
|
|
})
|
|
if storeId != "" {
|
|
store = fsw.storeIdToStore[storeId]
|
|
}
|
|
return
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) getDefaultStore() (store FilerStore) {
|
|
return fsw.defaultStore
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) GetName() string {
|
|
return fsw.getDefaultStore().GetName()
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) Initialize(configuration util.Configuration, prefix string) error {
|
|
return fsw.getDefaultStore().Initialize(configuration, prefix)
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) InsertEntry(ctx context.Context, entry *Entry) error {
|
|
actualStore := fsw.getActualStore(entry.FullPath)
|
|
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "insert").Inc()
|
|
start := time.Now()
|
|
defer func() {
|
|
stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "insert").Observe(time.Since(start).Seconds())
|
|
}()
|
|
|
|
filer_pb.BeforeEntrySerialization(entry.GetChunks())
|
|
if entry.Mime == "application/octet-stream" {
|
|
entry.Mime = ""
|
|
}
|
|
|
|
if err := fsw.handleUpdateToHardLinks(ctx, entry); err != nil {
|
|
return err
|
|
}
|
|
|
|
// glog.V(4).Infof("InsertEntry %s", entry.FullPath)
|
|
return actualStore.InsertEntry(ctx, entry)
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) UpdateEntry(ctx context.Context, entry *Entry) error {
|
|
actualStore := fsw.getActualStore(entry.FullPath)
|
|
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "update").Inc()
|
|
start := time.Now()
|
|
defer func() {
|
|
stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "update").Observe(time.Since(start).Seconds())
|
|
}()
|
|
|
|
filer_pb.BeforeEntrySerialization(entry.GetChunks())
|
|
if entry.Mime == "application/octet-stream" {
|
|
entry.Mime = ""
|
|
}
|
|
|
|
if err := fsw.handleUpdateToHardLinks(ctx, entry); err != nil {
|
|
return err
|
|
}
|
|
|
|
// glog.V(4).Infof("UpdateEntry %s", entry.FullPath)
|
|
return actualStore.UpdateEntry(ctx, entry)
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) FindEntry(ctx context.Context, fp util.FullPath) (entry *Entry, err error) {
|
|
actualStore := fsw.getActualStore(fp)
|
|
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "find").Inc()
|
|
start := time.Now()
|
|
defer func() {
|
|
stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "find").Observe(time.Since(start).Seconds())
|
|
}()
|
|
|
|
entry, err = actualStore.FindEntry(ctx, fp)
|
|
// glog.V(4).Infof("FindEntry %s: %v", fp, err)
|
|
if err != nil {
|
|
if fsw.CanDropWholeBucket() && strings.Contains(err.Error(), "Table") && strings.Contains(err.Error(), "doesn't exist") {
|
|
err = filer_pb.ErrNotFound
|
|
}
|
|
return nil, err
|
|
}
|
|
|
|
fsw.maybeReadHardLink(ctx, entry)
|
|
|
|
filer_pb.AfterEntryDeserialization(entry.GetChunks())
|
|
return
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) DeleteEntry(ctx context.Context, fp util.FullPath) (err error) {
|
|
actualStore := fsw.getActualStore(fp)
|
|
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "delete").Inc()
|
|
start := time.Now()
|
|
defer func() {
|
|
stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "delete").Observe(time.Since(start).Seconds())
|
|
}()
|
|
|
|
existingEntry, findErr := fsw.FindEntry(ctx, fp)
|
|
if findErr == filer_pb.ErrNotFound || existingEntry == nil {
|
|
return nil
|
|
}
|
|
if len(existingEntry.HardLinkId) != 0 {
|
|
// remove hard link
|
|
op := ctx.Value("OP")
|
|
if op != "MV" {
|
|
glog.V(4).InfofCtx(ctx, "DeleteHardLink %s", existingEntry.FullPath)
|
|
if err = fsw.DeleteHardLink(ctx, existingEntry.HardLinkId); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
|
|
// glog.V(4).Infof("DeleteEntry %s", fp)
|
|
return actualStore.DeleteEntry(ctx, fp)
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) DeleteOneEntry(ctx context.Context, existingEntry *Entry) (err error) {
|
|
actualStore := fsw.getActualStore(existingEntry.FullPath)
|
|
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "delete").Inc()
|
|
start := time.Now()
|
|
defer func() {
|
|
stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "delete").Observe(time.Since(start).Seconds())
|
|
}()
|
|
|
|
if len(existingEntry.HardLinkId) != 0 {
|
|
// remove hard link
|
|
op := ctx.Value("OP")
|
|
if op != "MV" {
|
|
glog.V(4).InfofCtx(ctx, "DeleteHardLink %s", existingEntry.FullPath)
|
|
if err = fsw.DeleteHardLink(ctx, existingEntry.HardLinkId); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
|
|
// glog.V(4).Infof("DeleteOneEntry %s", existingEntry.FullPath)
|
|
return actualStore.DeleteEntry(ctx, existingEntry.FullPath)
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) DeleteFolderChildren(ctx context.Context, fp util.FullPath) (err error) {
|
|
actualStore := fsw.getActualStore(fp + "/")
|
|
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "deleteFolderChildren").Inc()
|
|
start := time.Now()
|
|
defer func() {
|
|
stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "deleteFolderChildren").Observe(time.Since(start).Seconds())
|
|
}()
|
|
|
|
// glog.V(4).Infof("DeleteFolderChildren %s", fp)
|
|
return actualStore.DeleteFolderChildren(ctx, fp)
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc ListEachEntryFunc) (string, error) {
|
|
actualStore := fsw.getActualStore(dirPath + "/")
|
|
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "list").Inc()
|
|
start := time.Now()
|
|
defer func() {
|
|
stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "list").Observe(time.Since(start).Seconds())
|
|
}()
|
|
|
|
// glog.V(4).Infof("ListDirectoryEntries %s from %s limit %d", dirPath, startFileName, limit)
|
|
return actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *Entry) bool {
|
|
fsw.maybeReadHardLink(ctx, entry)
|
|
filer_pb.AfterEntryDeserialization(entry.GetChunks())
|
|
return eachEntryFunc(entry)
|
|
})
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc ListEachEntryFunc) (lastFileName string, err error) {
|
|
actualStore := fsw.getActualStore(dirPath + "/")
|
|
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "prefixList").Inc()
|
|
start := time.Now()
|
|
defer func() {
|
|
stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "prefixList").Observe(time.Since(start).Seconds())
|
|
}()
|
|
if limit > math.MaxInt32-1 {
|
|
limit = math.MaxInt32 - 1
|
|
}
|
|
// glog.V(4).Infof("ListDirectoryPrefixedEntries %s from %s prefix %s limit %d", dirPath, startFileName, prefix, limit)
|
|
adjustedEntryFunc := func(entry *Entry) bool {
|
|
fsw.maybeReadHardLink(ctx, entry)
|
|
filer_pb.AfterEntryDeserialization(entry.GetChunks())
|
|
return eachEntryFunc(entry)
|
|
}
|
|
lastFileName, err = actualStore.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, prefix, adjustedEntryFunc)
|
|
if err == ErrUnsupportedListDirectoryPrefixed {
|
|
lastFileName, err = fsw.prefixFilterEntries(ctx, dirPath, startFileName, includeStartFile, limit, prefix, adjustedEntryFunc)
|
|
}
|
|
return lastFileName, err
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) prefixFilterEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc ListEachEntryFunc) (lastFileName string, err error) {
|
|
actualStore := fsw.getActualStore(dirPath + "/")
|
|
|
|
if prefix == "" {
|
|
return actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, eachEntryFunc)
|
|
}
|
|
|
|
var notPrefixed []*Entry
|
|
lastFileName, err = actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *Entry) bool {
|
|
notPrefixed = append(notPrefixed, entry)
|
|
return true
|
|
})
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
count := int64(0)
|
|
for count < limit && len(notPrefixed) > 0 {
|
|
for _, entry := range notPrefixed {
|
|
if strings.HasPrefix(entry.Name(), prefix) {
|
|
count++
|
|
if !eachEntryFunc(entry) {
|
|
return
|
|
}
|
|
if count >= limit {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
if count < limit && lastFileName < prefix {
|
|
notPrefixed = notPrefixed[:0]
|
|
lastFileName, err = actualStore.ListDirectoryEntries(ctx, dirPath, lastFileName, false, limit, func(entry *Entry) bool {
|
|
notPrefixed = append(notPrefixed, entry)
|
|
return true
|
|
})
|
|
if err != nil {
|
|
return
|
|
}
|
|
} else {
|
|
break
|
|
}
|
|
}
|
|
return
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) BeginTransaction(ctx context.Context) (context.Context, error) {
|
|
return fsw.getDefaultStore().BeginTransaction(ctx)
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) CommitTransaction(ctx context.Context) error {
|
|
return fsw.getDefaultStore().CommitTransaction(ctx)
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) RollbackTransaction(ctx context.Context) error {
|
|
return fsw.getDefaultStore().RollbackTransaction(ctx)
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) Shutdown() {
|
|
fsw.getDefaultStore().Shutdown()
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) KvPut(ctx context.Context, key []byte, value []byte) (err error) {
|
|
return fsw.getDefaultStore().KvPut(ctx, key, value)
|
|
}
|
|
func (fsw *FilerStoreWrapper) KvGet(ctx context.Context, key []byte) (value []byte, err error) {
|
|
return fsw.getDefaultStore().KvGet(ctx, key)
|
|
}
|
|
func (fsw *FilerStoreWrapper) KvDelete(ctx context.Context, key []byte) (err error) {
|
|
return fsw.getDefaultStore().KvDelete(ctx, key)
|
|
}
|
|
|
|
func (fsw *FilerStoreWrapper) Debug(writer io.Writer) {
|
|
if debuggable, ok := fsw.getDefaultStore().(Debuggable); ok {
|
|
debuggable.Debug(writer)
|
|
}
|
|
}
|