seaweedfs/weed/filer/filerstore.go

326 lines
11 KiB
Go
Raw Normal View History

2020-09-01 07:21:19 +00:00
package filer
2018-05-26 06:27:06 +00:00
import (
2019-03-15 22:55:34 +00:00
"context"
"errors"
2020-12-11 03:55:28 +00:00
"github.com/chrislusf/seaweedfs/weed/glog"
2020-12-18 10:55:00 +00:00
"github.com/viant/ptrie"
"strings"
2019-06-22 19:23:25 +00:00
"time"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
2019-06-22 19:23:25 +00:00
"github.com/chrislusf/seaweedfs/weed/stats"
"github.com/chrislusf/seaweedfs/weed/util"
)
2018-05-26 06:27:06 +00:00
var (
ErrUnsupportedListDirectoryPrefixed = errors.New("unsupported directory prefix listing")
ErrKvNotImplemented = errors.New("kv not implemented yet")
ErrKvNotFound = errors.New("kv: not found")
_ = VirtualFilerStore(&FilerStoreWrapper{})
)
2018-05-26 06:27:06 +00:00
type FilerStore interface {
// GetName gets the name to locate the configuration in filer.toml file
GetName() string
2018-06-17 20:01:57 +00:00
// Initialize initializes the file store
Initialize(configuration util.Configuration, prefix string) error
2019-03-15 22:55:34 +00:00
InsertEntry(context.Context, *Entry) error
UpdateEntry(context.Context, *Entry) (err error)
2020-09-24 10:06:44 +00:00
// err == filer_pb.ErrNotFound if not found
2020-03-23 07:01:34 +00:00
FindEntry(context.Context, util.FullPath) (entry *Entry, err error)
DeleteEntry(context.Context, util.FullPath) (err error)
DeleteFolderChildren(context.Context, util.FullPath) (err error)
ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int) ([]*Entry, error)
2020-08-05 17:19:16 +00:00
ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int, prefix string) ([]*Entry, error)
BeginTransaction(ctx context.Context) (context.Context, error)
CommitTransaction(ctx context.Context) error
RollbackTransaction(ctx context.Context) error
2020-03-15 03:30:26 +00:00
KvPut(ctx context.Context, key []byte, value []byte) (err error)
KvGet(ctx context.Context, key []byte) (value []byte, err error)
KvDelete(ctx context.Context, key []byte) (err error)
2020-03-15 03:30:26 +00:00
Shutdown()
}
2020-09-24 10:06:44 +00:00
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)
2020-09-24 10:06:44 +00:00
}
type FilerStoreWrapper struct {
defaultStore FilerStore
pathToStore ptrie.Trie
storeIdToStore map[string]FilerStore
}
2019-05-29 04:29:07 +00:00
func NewFilerStoreWrapper(store FilerStore) *FilerStoreWrapper {
if innerStore, ok := store.(*FilerStoreWrapper); ok {
return innerStore
}
return &FilerStoreWrapper{
defaultStore: store,
pathToStore: ptrie.New(),
storeIdToStore: make(map[string]FilerStore),
}
}
func (fsw *FilerStoreWrapper) AddPathSpecificStore(path string, storeId string, store FilerStore) {
fsw.storeIdToStore[storeId] = store
err := fsw.pathToStore.Put([]byte(path), storeId)
if err != nil {
glog.Fatalf("put path specific store: %v", err)
}
2020-12-18 10:55:00 +00:00
}
func (fsw *FilerStoreWrapper) getActualStore(path util.FullPath) (store FilerStore) {
store = fsw.defaultStore
2020-12-18 10:55:00 +00:00
if path == "" {
return
2020-12-18 10:55:00 +00:00
}
var storeId string
fsw.pathToStore.MatchPrefix([]byte(path), func(key []byte, value interface{}) bool {
storeId = value.(string)
return false
})
if storeId != "" {
store = fsw.storeIdToStore[storeId]
2020-12-18 10:55:00 +00:00
}
return
2020-12-18 10:35:45 +00:00
}
func (fsw *FilerStoreWrapper) GetName() string {
2020-12-18 10:35:45 +00:00
return fsw.getActualStore("").GetName()
}
func (fsw *FilerStoreWrapper) Initialize(configuration util.Configuration, prefix string) error {
2020-12-18 10:35:45 +00:00
return fsw.getActualStore("").Initialize(configuration, prefix)
}
func (fsw *FilerStoreWrapper) InsertEntry(ctx context.Context, entry *Entry) error {
2020-12-18 10:57:49 +00:00
actualStore := fsw.getActualStore(entry.FullPath)
2020-12-18 10:55:00 +00:00
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "insert").Inc()
2019-06-22 19:23:25 +00:00
start := time.Now()
2019-06-23 03:05:25 +00:00
defer func() {
2020-12-18 10:55:00 +00:00
stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "insert").Observe(time.Since(start).Seconds())
2019-06-23 03:05:25 +00:00
}()
2019-06-22 19:23:25 +00:00
filer_pb.BeforeEntrySerialization(entry.Chunks)
if entry.Mime == "application/octet-stream" {
entry.Mime = ""
}
2020-09-24 10:06:44 +00:00
if err := fsw.handleUpdateToHardLinks(ctx, entry); err != nil {
return err
2020-09-24 10:06:44 +00:00
}
2020-12-11 03:55:28 +00:00
glog.V(4).Infof("InsertEntry %s", entry.FullPath)
2020-12-18 10:55:00 +00:00
return actualStore.InsertEntry(ctx, entry)
}
func (fsw *FilerStoreWrapper) UpdateEntry(ctx context.Context, entry *Entry) error {
2020-12-18 10:57:49 +00:00
actualStore := fsw.getActualStore(entry.FullPath)
2020-12-18 10:55:00 +00:00
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "update").Inc()
2019-06-22 19:23:25 +00:00
start := time.Now()
2019-06-23 03:05:25 +00:00
defer func() {
2020-12-18 10:55:00 +00:00
stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "update").Observe(time.Since(start).Seconds())
2019-06-23 03:05:25 +00:00
}()
2019-06-22 19:23:25 +00:00
filer_pb.BeforeEntrySerialization(entry.Chunks)
if entry.Mime == "application/octet-stream" {
entry.Mime = ""
}
2020-09-24 10:06:44 +00:00
if err := fsw.handleUpdateToHardLinks(ctx, entry); err != nil {
return err
2020-09-24 10:06:44 +00:00
}
2020-12-11 03:55:28 +00:00
glog.V(4).Infof("UpdateEntry %s", entry.FullPath)
2020-12-18 10:55:00 +00:00
return actualStore.UpdateEntry(ctx, entry)
}
2020-03-23 07:01:34 +00:00
func (fsw *FilerStoreWrapper) FindEntry(ctx context.Context, fp util.FullPath) (entry *Entry, err error) {
2020-12-18 10:57:49 +00:00
actualStore := fsw.getActualStore(fp)
2020-12-18 10:55:00 +00:00
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "find").Inc()
2019-06-22 19:23:25 +00:00
start := time.Now()
2019-06-23 03:05:25 +00:00
defer func() {
2020-12-18 10:55:00 +00:00
stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "find").Observe(time.Since(start).Seconds())
2019-06-23 03:05:25 +00:00
}()
2019-06-22 19:23:25 +00:00
2020-12-11 03:55:28 +00:00
glog.V(4).Infof("FindEntry %s", fp)
2020-12-18 10:55:00 +00:00
entry, err = actualStore.FindEntry(ctx, fp)
2019-05-17 09:28:20 +00:00
if err != nil {
return nil, err
}
2020-09-24 10:06:44 +00:00
fsw.maybeReadHardLink(ctx, entry)
filer_pb.AfterEntryDeserialization(entry.Chunks)
return
}
2020-03-23 07:01:34 +00:00
func (fsw *FilerStoreWrapper) DeleteEntry(ctx context.Context, fp util.FullPath) (err error) {
2020-12-18 10:57:49 +00:00
actualStore := fsw.getActualStore(fp)
2020-12-18 10:55:00 +00:00
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "delete").Inc()
2019-06-22 19:23:25 +00:00
start := time.Now()
2019-06-23 03:05:25 +00:00
defer func() {
2020-12-18 10:55:00 +00:00
stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "delete").Observe(time.Since(start).Seconds())
2019-06-23 03:05:25 +00:00
}()
2019-06-22 19:23:25 +00:00
2020-09-24 10:06:44 +00:00
existingEntry, findErr := fsw.FindEntry(ctx, fp)
if findErr == filer_pb.ErrNotFound {
return nil
}
2020-09-24 18:11:42 +00:00
if len(existingEntry.HardLinkId) != 0 {
2020-09-24 10:06:44 +00:00
// remove hard link
2020-12-11 03:55:28 +00:00
glog.V(4).Infof("DeleteHardLink %s", existingEntry.FullPath)
2020-09-24 10:06:44 +00:00
if err = fsw.DeleteHardLink(ctx, existingEntry.HardLinkId); err != nil {
return err
}
}
2020-12-11 03:55:28 +00:00
glog.V(4).Infof("DeleteEntry %s", fp)
2020-12-18 10:55:00 +00:00
return actualStore.DeleteEntry(ctx, fp)
}
func (fsw *FilerStoreWrapper) DeleteOneEntry(ctx context.Context, existingEntry *Entry) (err error) {
2020-12-18 10:57:49 +00:00
actualStore := fsw.getActualStore(existingEntry.FullPath)
2020-12-18 10:55:00 +00:00
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "delete").Inc()
start := time.Now()
defer func() {
2020-12-18 10:55:00 +00:00
stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "delete").Observe(time.Since(start).Seconds())
}()
if len(existingEntry.HardLinkId) != 0 {
// remove hard link
2020-12-11 03:55:28 +00:00
glog.V(4).Infof("DeleteHardLink %s", existingEntry.FullPath)
if err = fsw.DeleteHardLink(ctx, existingEntry.HardLinkId); err != nil {
return err
}
}
2020-12-12 11:33:57 +00:00
glog.V(4).Infof("DeleteOneEntry %s", existingEntry.FullPath)
2020-12-18 10:55:00 +00:00
return actualStore.DeleteEntry(ctx, existingEntry.FullPath)
}
2020-03-23 07:01:34 +00:00
func (fsw *FilerStoreWrapper) DeleteFolderChildren(ctx context.Context, fp util.FullPath) (err error) {
2020-12-18 10:57:49 +00:00
actualStore := fsw.getActualStore(fp)
2020-12-18 10:55:00 +00:00
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "deleteFolderChildren").Inc()
start := time.Now()
defer func() {
2020-12-18 10:55:00 +00:00
stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "deleteFolderChildren").Observe(time.Since(start).Seconds())
}()
2020-12-11 03:55:28 +00:00
glog.V(4).Infof("DeleteFolderChildren %s", fp)
2020-12-18 10:55:00 +00:00
return actualStore.DeleteFolderChildren(ctx, fp)
}
2020-03-23 07:01:34 +00:00
func (fsw *FilerStoreWrapper) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int) ([]*Entry, error) {
2020-12-18 10:57:49 +00:00
actualStore := fsw.getActualStore(dirPath)
2020-12-18 10:55:00 +00:00
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "list").Inc()
2019-06-22 19:23:25 +00:00
start := time.Now()
2019-06-23 03:05:25 +00:00
defer func() {
2020-12-18 10:55:00 +00:00
stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "list").Observe(time.Since(start).Seconds())
2019-06-23 03:05:25 +00:00
}()
2019-06-22 19:23:25 +00:00
2020-12-11 03:55:28 +00:00
glog.V(4).Infof("ListDirectoryEntries %s from %s limit %d", dirPath, startFileName, limit)
2020-12-18 10:55:00 +00:00
entries, err := actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit)
if err != nil {
return nil, err
}
for _, entry := range entries {
2020-09-24 10:06:44 +00:00
fsw.maybeReadHardLink(ctx, entry)
filer_pb.AfterEntryDeserialization(entry.Chunks)
}
return entries, err
}
2020-08-05 18:38:00 +00:00
func (fsw *FilerStoreWrapper) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int, prefix string) ([]*Entry, error) {
2020-12-18 10:57:49 +00:00
actualStore := fsw.getActualStore(dirPath)
2020-12-18 10:55:00 +00:00
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "prefixList").Inc()
2020-08-05 18:38:00 +00:00
start := time.Now()
defer func() {
2020-12-18 10:55:00 +00:00
stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "prefixList").Observe(time.Since(start).Seconds())
2020-08-05 18:38:00 +00:00
}()
2020-12-11 03:55:28 +00:00
glog.V(4).Infof("ListDirectoryPrefixedEntries %s from %s prefix %s limit %d", dirPath, startFileName, prefix, limit)
2020-12-18 10:55:00 +00:00
entries, err := actualStore.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, prefix)
2020-08-31 16:55:18 +00:00
if err == ErrUnsupportedListDirectoryPrefixed {
2020-08-31 17:39:24 +00:00
entries, err = fsw.prefixFilterEntries(ctx, dirPath, startFileName, includeStartFile, limit, prefix)
}
if err != nil {
2020-08-05 18:38:00 +00:00
return nil, err
}
for _, entry := range entries {
2020-09-24 10:06:44 +00:00
fsw.maybeReadHardLink(ctx, entry)
2020-08-05 18:38:00 +00:00
filer_pb.AfterEntryDeserialization(entry.Chunks)
}
return entries, nil
}
2020-08-31 17:39:24 +00:00
func (fsw *FilerStoreWrapper) prefixFilterEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int, prefix string) (entries []*Entry, err error) {
2020-12-18 10:57:49 +00:00
actualStore := fsw.getActualStore(dirPath)
2020-12-18 10:55:00 +00:00
entries, err = actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit)
2020-08-31 17:39:24 +00:00
if err != nil {
return nil, err
}
if prefix == "" {
return
}
count := 0
var lastFileName string
notPrefixed := entries
entries = nil
for count < limit && len(notPrefixed) > 0 {
for _, entry := range notPrefixed {
lastFileName = entry.Name()
if strings.HasPrefix(entry.Name(), prefix) {
count++
entries = append(entries, entry)
2020-08-31 18:28:03 +00:00
if count >= limit {
break
}
2020-08-31 17:39:24 +00:00
}
}
if count < limit {
2020-12-18 10:55:00 +00:00
notPrefixed, err = actualStore.ListDirectoryEntries(ctx, dirPath, lastFileName, false, limit)
2020-08-31 17:39:24 +00:00
if err != nil {
return
}
}
}
return
}
func (fsw *FilerStoreWrapper) BeginTransaction(ctx context.Context) (context.Context, error) {
2020-12-18 10:35:45 +00:00
return fsw.getActualStore("").BeginTransaction(ctx)
}
func (fsw *FilerStoreWrapper) CommitTransaction(ctx context.Context) error {
2020-12-18 10:35:45 +00:00
return fsw.getActualStore("").CommitTransaction(ctx)
}
func (fsw *FilerStoreWrapper) RollbackTransaction(ctx context.Context) error {
2020-12-18 10:35:45 +00:00
return fsw.getActualStore("").RollbackTransaction(ctx)
}
2020-03-15 03:30:26 +00:00
func (fsw *FilerStoreWrapper) Shutdown() {
2020-12-18 10:35:45 +00:00
fsw.getActualStore("").Shutdown()
2020-03-15 03:30:26 +00:00
}
func (fsw *FilerStoreWrapper) KvPut(ctx context.Context, key []byte, value []byte) (err error) {
2020-12-18 10:35:45 +00:00
return fsw.getActualStore("").KvPut(ctx, key, value)
}
func (fsw *FilerStoreWrapper) KvGet(ctx context.Context, key []byte) (value []byte, err error) {
2020-12-18 10:35:45 +00:00
return fsw.getActualStore("").KvGet(ctx, key)
}
func (fsw *FilerStoreWrapper) KvDelete(ctx context.Context, key []byte) (err error) {
2020-12-18 10:35:45 +00:00
return fsw.getActualStore("").KvDelete(ctx, key)
}