diff --git a/weed/filer/filer.go b/weed/filer/filer.go index 4595c55f4..589fd6066 100644 --- a/weed/filer/filer.go +++ b/weed/filer/filer.go @@ -165,7 +165,10 @@ func (f *Filer) ListExistingPeerUpdates(ctx context.Context) (existingNodes []*m } func (f *Filer) SetStore(store FilerStore) (isFresh bool) { - f.Store = NewFilerStoreWrapper(store) + fsw := NewFilerStoreWrapper(store) + // f.FilerConf is swapped on filer.conf reloads, so resolve it per call. + fsw.filerConfFn = func() *FilerConf { return f.FilerConf } + f.Store = fsw return f.setOrLoadFilerStoreSignature(store) } diff --git a/weed/filer/filer_conf.go b/weed/filer/filer_conf.go index f80c63f63..639265d07 100644 --- a/weed/filer/filer_conf.go +++ b/weed/filer/filer_conf.go @@ -5,6 +5,7 @@ import ( "context" "fmt" "io" + "strings" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/wdclient" @@ -230,9 +231,28 @@ func ClonePathConf(src *filer_pb.FilerConf_PathConf) *filer_pb.FilerConf_PathCon Worm: src.Worm, WormGracePeriodSeconds: src.WormGracePeriodSeconds, WormRetentionTimeSeconds: src.WormRetentionTimeSeconds, + InodeIndex: src.InodeIndex, } } +// InodeIndexActiveUnder reports whether entries inside dirPath may carry inode +// index rows: either dirPath itself is covered by an inode_index rule, or such +// a rule applies to a subtree inside dirPath. +func (fc *FilerConf) InodeIndexActiveUnder(dirPath string) bool { + if fc.MatchStorageRule(dirPath).InodeIndex { + return true + } + active := false + fc.rules.Walk(func(key []byte, value *filer_pb.FilerConf_PathConf) bool { + if value.InodeIndex && strings.HasPrefix(string(key), dirPath) { + active = true + return false + } + return true + }) + return active +} + // ApplyBucketQuotaReadOnly sets read-only when usedSize exceeds quota and clears it // once back under, reporting whether the flag changed. A non-positive quota is left // untouched so a manually locked bucket is never reopened. @@ -292,6 +312,7 @@ func mergePathConf(a, b *filer_pb.FilerConf_PathConf) { a.DataNode = util.Nvl(b.DataNode, a.DataNode) a.DisableChunkDeletion = b.DisableChunkDeletion || a.DisableChunkDeletion a.Worm = b.Worm || a.Worm + a.InodeIndex = b.InodeIndex || a.InodeIndex if b.WormRetentionTimeSeconds > 0 { a.WormRetentionTimeSeconds = b.WormRetentionTimeSeconds } diff --git a/weed/filer/filer_conf_test.go b/weed/filer/filer_conf_test.go index e0d430a98..03b1487c4 100644 --- a/weed/filer/filer_conf_test.go +++ b/weed/filer/filer_conf_test.go @@ -71,6 +71,7 @@ func TestClonePathConf(t *testing.T) { Worm: true, WormGracePeriodSeconds: 3600, WormRetentionTimeSeconds: 86400, + InodeIndex: true, } clone := ClonePathConf(src) diff --git a/weed/filer/filer_inode_index.go b/weed/filer/filer_inode_index.go new file mode 100644 index 000000000..2608fad3b --- /dev/null +++ b/weed/filer/filer_inode_index.go @@ -0,0 +1,335 @@ +package filer + +import ( + "context" + "encoding/binary" + "encoding/json" + "sort" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +const inodeIndexKeyPrefix = "filer.inode.path." +const InodeIndexInitialGeneration uint64 = 1 + +type inodeIndexEntry struct { + path util.FullPath + inode uint64 +} + +// InodeIndexRecord is the inode→path reverse-lookup row consumed by the NFS +// gateway to resolve filehandles. Rows are maintained only for paths covered +// by an inode_index rule in filer.conf (see shell nfs.enable); everywhere else +// the filer writes nothing. +type InodeIndexRecord struct { + Generation uint64 `json:"generation,omitempty"` + Paths []string `json:"paths,omitempty"` +} + +func InodeIndexKey(inode uint64) []byte { + key := make([]byte, len(inodeIndexKeyPrefix)+8) + copy(key, inodeIndexKeyPrefix) + binary.BigEndian.PutUint64(key[len(inodeIndexKeyPrefix):], inode) + return key +} + +func DecodeInodeIndexRecord(value []byte) (*InodeIndexRecord, error) { + if len(value) == 0 { + return &InodeIndexRecord{}, nil + } + + // The first foundation slice stored the current path as raw bytes. Keep that + // format readable so existing records are transparently upgraded on write. + if value[0] != '{' { + record := &InodeIndexRecord{Generation: InodeIndexInitialGeneration} + record.AddPath(util.FullPath(value)) + return record, nil + } + + record := &InodeIndexRecord{} + if err := json.Unmarshal(value, record); err != nil { + return nil, err + } + record.normalize() + return record, nil +} + +func (record *InodeIndexRecord) Encode() ([]byte, error) { + record.normalize() + return json.Marshal(record) +} + +func (record *InodeIndexRecord) normalize() { + if len(record.Paths) == 0 { + return + } + if record.Generation == 0 { + record.Generation = InodeIndexInitialGeneration + } + + sanitized := make([]string, 0, len(record.Paths)) + for _, path := range record.Paths { + if path == "" { + continue + } + sanitized = append(sanitized, path) + } + if len(sanitized) == 0 { + record.Paths = nil + return + } + + sort.Strings(sanitized) + deduped := sanitized[:1] + for _, path := range sanitized[1:] { + if path == deduped[len(deduped)-1] { + continue + } + deduped = append(deduped, path) + } + record.Paths = deduped +} + +func (record *InodeIndexRecord) AddPath(path util.FullPath) bool { + if path == "" { + return false + } + record.normalize() + target := string(path) + index := sort.SearchStrings(record.Paths, target) + if index < len(record.Paths) && record.Paths[index] == target { + return false + } + record.Paths = append(record.Paths, "") + copy(record.Paths[index+1:], record.Paths[index:]) + record.Paths[index] = target + return true +} + +func (record *InodeIndexRecord) RemovePath(path util.FullPath) bool { + if len(record.Paths) == 0 || path == "" { + return false + } + record.normalize() + target := string(path) + index := sort.SearchStrings(record.Paths, target) + if index >= len(record.Paths) || record.Paths[index] != target { + return false + } + record.Paths = append(record.Paths[:index], record.Paths[index+1:]...) + if len(record.Paths) == 0 { + record.Paths = nil + } + return true +} + +func (record *InodeIndexRecord) CanonicalPath() util.FullPath { + record.normalize() + if len(record.Paths) == 0 { + return "" + } + return util.FullPath(record.Paths[0]) +} + +func (record *InodeIndexRecord) FullPaths() []util.FullPath { + record.normalize() + if len(record.Paths) == 0 { + return nil + } + paths := make([]util.FullPath, 0, len(record.Paths)) + for _, path := range record.Paths { + paths = append(paths, util.FullPath(path)) + } + return paths +} + +// inodeIndexInScope reports whether the entry at path gets an inode index row. +// The index is opt-in per filer.conf inode_index location rules; with no +// matching rule (the default) nothing is ever written. +func (fsw *FilerStoreWrapper) inodeIndexInScope(path util.FullPath) bool { + if fsw.filerConfFn == nil { + return false + } + fc := fsw.filerConfFn() + if fc == nil { + return false + } + return fc.MatchStorageRule(string(path)).InodeIndex +} + +// inodeIndexUnder reports whether any entry inside dirPath may carry an inode +// index row, so DeleteFolderChildren can skip the housekeeping walk entirely +// when the subtree does not intersect any inode_index rule. +func (fsw *FilerStoreWrapper) inodeIndexUnder(dirPath util.FullPath) bool { + if fsw.filerConfFn == nil { + return false + } + fc := fsw.filerConfFn() + if fc == nil { + return false + } + return fc.InodeIndexActiveUnder(string(dirPath)) +} + +func (fsw *FilerStoreWrapper) lookupInodeIndex(ctx context.Context, inode uint64) (*InodeIndexRecord, error) { + if inode == 0 { + return nil, ErrKvNotFound + } + + value, err := fsw.KvGet(ctx, InodeIndexKey(inode)) + if err != nil { + return nil, err + } + + return DecodeInodeIndexRecord(value) +} + +func (fsw *FilerStoreWrapper) storeInodeIndex(ctx context.Context, path util.FullPath, inode uint64) error { + if inode == 0 || path == "" { + return nil + } + + record, err := fsw.lookupInodeIndex(ctx, inode) + if err != nil { + if err != ErrKvNotFound { + return err + } + record = &InodeIndexRecord{Generation: InodeIndexInitialGeneration} + } + record.AddPath(path) + + value, err := record.Encode() + if err != nil { + return err + } + return fsw.KvPut(ctx, InodeIndexKey(inode), value) +} + +func (fsw *FilerStoreWrapper) lookupInodePath(ctx context.Context, inode uint64) (util.FullPath, error) { + record, err := fsw.lookupInodeIndex(ctx, inode) + if err != nil { + return "", err + } + + path := record.CanonicalPath() + if path == "" { + return "", ErrKvNotFound + } + return path, nil +} + +func (fsw *FilerStoreWrapper) lookupInodePaths(ctx context.Context, inode uint64) ([]util.FullPath, error) { + record, err := fsw.lookupInodeIndex(ctx, inode) + if err != nil { + return nil, err + } + + paths := record.FullPaths() + if len(paths) == 0 { + return nil, ErrKvNotFound + } + return paths, nil +} + +func (fsw *FilerStoreWrapper) removePathFromInodeIndex(ctx context.Context, path util.FullPath, inode uint64) error { + if inode == 0 || path == "" { + return nil + } + + record, err := fsw.lookupInodeIndex(ctx, inode) + if err != nil { + if err == ErrKvNotFound { + return nil + } + return err + } + + if !record.RemovePath(path) { + return nil + } + if len(record.Paths) == 0 { + return fsw.KvDelete(ctx, InodeIndexKey(inode)) + } + + value, err := record.Encode() + if err != nil { + return err + } + return fsw.KvPut(ctx, InodeIndexKey(inode), value) +} + +func (fsw *FilerStoreWrapper) collectInodeIndexEntries(ctx context.Context, dirPath util.FullPath) ([]inodeIndexEntry, error) { + if !fsw.inodeIndexUnder(dirPath) { + return nil, nil + } + // Honor caller cancellation during the walk: a DeleteFolderChildren on a + // pathological directory could otherwise loop indefinitely gathering + // entries even after the client has given up, turning into a DoS vector. + // If the walk is aborted, the caller treats the index cleanup as + // best-effort and drops the partial result. + var collected []inodeIndexEntry + if err := fsw.collectInodeIndexEntriesRecursive(ctx, dirPath, &collected); err != nil { + return nil, err + } + return collected, nil +} + +func (fsw *FilerStoreWrapper) collectInodeIndexEntriesRecursive(ctx context.Context, dirPath util.FullPath, collected *[]inodeIndexEntry) error { + actualStore := fsw.getActualStore(dirPath + "/") + + lastFileName := "" + includeStartFile := false + for { + page := make([]*Entry, 0, PaginationSize) + nextLastFileName, err := actualStore.ListDirectoryEntries(ctx, dirPath, lastFileName, includeStartFile, PaginationSize, func(entry *Entry) (bool, error) { + page = append(page, entry) + return true, nil + }) + if err != nil { + return err + } + + for _, entry := range page { + if entry.Attr.Inode != 0 { + *collected = append(*collected, inodeIndexEntry{path: entry.FullPath, inode: entry.Attr.Inode}) + } + if entry.IsDirectory() { + if err := fsw.collectInodeIndexEntriesRecursive(ctx, entry.FullPath, collected); err != nil { + return err + } + } + } + + if len(page) < PaginationSize { + return nil + } + lastFileName = nextLastFileName + includeStartFile = false + } +} + +// recordInodeIndexWrite updates the inode→path secondary index after the +// primary store mutation has already succeeded. The index is best-effort: a +// failure here must not surface as an operation error, because the caller +// would then observe a failed create/update even though the entry was +// persisted, and a retry cannot heal the index (DeleteEntry exits early once +// the entry is gone). We log and let later writes rebuild the record. +func (fsw *FilerStoreWrapper) recordInodeIndexWrite(ctx context.Context, op string, path util.FullPath, inode uint64) { + if inode == 0 || path == "" || !fsw.inodeIndexInScope(path) { + return + } + if err := fsw.storeInodeIndex(ctx, path, inode); err != nil { + glog.WarningfCtx(ctx, "%s: update inode index for %s (inode %d): %v", op, path, inode, err) + } +} + +// recordInodeIndexRemoval mirrors recordInodeIndexWrite for removals. +func (fsw *FilerStoreWrapper) recordInodeIndexRemoval(ctx context.Context, op string, path util.FullPath, inode uint64) { + if inode == 0 || path == "" || !fsw.inodeIndexInScope(path) { + return + } + if err := fsw.removePathFromInodeIndex(ctx, path, inode); err != nil { + glog.WarningfCtx(ctx, "%s: clear inode index for %s (inode %d): %v", op, path, inode, err) + } +} diff --git a/weed/filer/filer_inode_index_test.go b/weed/filer/filer_inode_index_test.go new file mode 100644 index 000000000..9172dd383 --- /dev/null +++ b/weed/filer/filer_inode_index_test.go @@ -0,0 +1,301 @@ +package filer + +import ( + "context" + "os" + "strings" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func inodeIndexConf(prefixes ...string) *FilerConf { + fc := NewFilerConf() + for _, prefix := range prefixes { + _ = fc.SetLocationConf(&filer_pb.FilerConf_PathConf{LocationPrefix: prefix, InodeIndex: true}) + } + return fc +} + +func newInodeIndexWrapper(prefixes ...string) (*FilerStoreWrapper, *stubFilerStore) { + store := newStubFilerStore() + wrapper := NewFilerStoreWrapper(store) + fc := inodeIndexConf(prefixes...) + wrapper.filerConfFn = func() *FilerConf { return fc } + return wrapper, store +} + +func countInodeIndexRows(store *stubFilerStore) int { + store.mu.Lock() + defer store.mu.Unlock() + count := 0 + for key := range store.kv { + if strings.HasPrefix(key, inodeIndexKeyPrefix) { + count++ + } + } + return count +} + +func TestFilerStoreWrapperMaintainsInodeIndexLifecycle(t *testing.T) { + wrapper, _ := newInodeIndexWrapper("/") + ctx := context.Background() + + created := &Entry{ + FullPath: util.FullPath("/docs/report.txt"), + Attr: Attr{ + Mode: 0o644, + Inode: 42, + }, + } + + require.NoError(t, wrapper.InsertEntry(ctx, created)) + path, err := wrapper.lookupInodePath(ctx, created.Attr.Inode) + require.NoError(t, err) + assert.Equal(t, created.FullPath, path) + paths, err := wrapper.lookupInodePaths(ctx, created.Attr.Inode) + require.NoError(t, err) + assert.Equal(t, []util.FullPath{created.FullPath}, paths) + record, err := wrapper.lookupInodeIndex(ctx, created.Attr.Inode) + require.NoError(t, err) + assert.Equal(t, InodeIndexInitialGeneration, record.Generation) + + updated := &Entry{ + FullPath: util.FullPath("/docs/report.txt"), + Attr: Attr{ + Mode: 0o600, + Inode: 42, + }, + } + require.NoError(t, wrapper.UpdateEntry(ctx, updated)) + path, err = wrapper.lookupInodePath(ctx, updated.Attr.Inode) + require.NoError(t, err) + assert.Equal(t, updated.FullPath, path) + + require.NoError(t, wrapper.DeleteEntry(ctx, created.FullPath)) + _, err = wrapper.lookupInodePath(ctx, created.Attr.Inode) + require.ErrorIs(t, err, ErrKvNotFound) +} + +func TestFilerStoreWrapperMaintainsMultiplePathsPerInode(t *testing.T) { + wrapper, _ := newInodeIndexWrapper("/") + ctx := context.Background() + inode := uint64(88) + hardLinkId := NewHardLinkId() + + require.NoError(t, wrapper.InsertEntry(ctx, &Entry{ + FullPath: util.FullPath("/links/b.txt"), + Attr: Attr{ + Mode: 0o644, + Inode: inode, + }, + HardLinkId: hardLinkId, + HardLinkCounter: 2, + })) + require.NoError(t, wrapper.InsertEntry(ctx, &Entry{ + FullPath: util.FullPath("/links/a.txt"), + Attr: Attr{ + Mode: 0o644, + Inode: inode, + }, + HardLinkId: hardLinkId, + HardLinkCounter: 2, + })) + + paths, err := wrapper.lookupInodePaths(ctx, inode) + require.NoError(t, err) + assert.Equal(t, []util.FullPath{"/links/a.txt", "/links/b.txt"}, paths) + record, err := wrapper.lookupInodeIndex(ctx, inode) + require.NoError(t, err) + assert.Equal(t, InodeIndexInitialGeneration, record.Generation) + + path, err := wrapper.lookupInodePath(ctx, inode) + require.NoError(t, err) + assert.Equal(t, util.FullPath("/links/a.txt"), path) + + require.NoError(t, wrapper.DeleteEntry(ctx, util.FullPath("/links/a.txt"))) + + paths, err = wrapper.lookupInodePaths(ctx, inode) + require.NoError(t, err) + assert.Equal(t, []util.FullPath{"/links/b.txt"}, paths) + + path, err = wrapper.lookupInodePath(ctx, inode) + require.NoError(t, err) + assert.Equal(t, util.FullPath("/links/b.txt"), path) +} + +func TestFilerStoreWrapperUpgradesLegacySinglePathInodeIndexRecords(t *testing.T) { + wrapper := NewFilerStoreWrapper(newStubFilerStore()) + ctx := context.Background() + inode := uint64(91) + + require.NoError(t, wrapper.KvPut(ctx, InodeIndexKey(inode), []byte("/legacy/path.txt"))) + + path, err := wrapper.lookupInodePath(ctx, inode) + require.NoError(t, err) + assert.Equal(t, util.FullPath("/legacy/path.txt"), path) + + paths, err := wrapper.lookupInodePaths(ctx, inode) + require.NoError(t, err) + assert.Equal(t, []util.FullPath{"/legacy/path.txt"}, paths) + + require.NoError(t, wrapper.storeInodeIndex(ctx, util.FullPath("/legacy/second.txt"), inode)) + + paths, err = wrapper.lookupInodePaths(ctx, inode) + require.NoError(t, err) + assert.Equal(t, []util.FullPath{"/legacy/path.txt", "/legacy/second.txt"}, paths) + + value, err := wrapper.KvGet(ctx, InodeIndexKey(inode)) + require.NoError(t, err) + assert.JSONEq(t, `{"generation":1,"paths":["/legacy/path.txt","/legacy/second.txt"]}`, string(value)) +} + +func TestFilerStoreWrapperKeepsInodeIndexWhenDeleteArrivesAfterRenameInsert(t *testing.T) { + wrapper, _ := newInodeIndexWrapper("/") + ctx := context.Background() + inode := uint64(77) + + require.NoError(t, wrapper.InsertEntry(ctx, &Entry{ + FullPath: util.FullPath("/old/name.txt"), + Attr: Attr{ + Mode: 0o644, + Inode: inode, + }, + })) + require.NoError(t, wrapper.InsertEntry(ctx, &Entry{ + FullPath: util.FullPath("/new/name.txt"), + Attr: Attr{ + Mode: 0o644, + Inode: inode, + }, + })) + require.NoError(t, wrapper.DeleteEntry(ctx, util.FullPath("/old/name.txt"))) + + path, err := wrapper.lookupInodePath(ctx, inode) + require.NoError(t, err) + assert.Equal(t, util.FullPath("/new/name.txt"), path) + + paths, err := wrapper.lookupInodePaths(ctx, inode) + require.NoError(t, err) + assert.Equal(t, []util.FullPath{"/new/name.txt"}, paths) +} + +func TestInodeIndexDisabledWritesNothing(t *testing.T) { + // No filer.conf wired at all, and a conf with no inode_index rules: both + // must leave the KV store free of index rows. + t.Run("no conf", func(t *testing.T) { + store := newStubFilerStore() + runInodeIndexDisabledScenario(t, NewFilerStoreWrapper(store), store) + }) + t.Run("conf without rules", func(t *testing.T) { + wrapper, store := newInodeIndexWrapper() + runInodeIndexDisabledScenario(t, wrapper, store) + }) +} + +func runInodeIndexDisabledScenario(t *testing.T, wrapper *FilerStoreWrapper, store *stubFilerStore) { + ctx := context.Background() + entry := &Entry{ + FullPath: util.FullPath("/docs/report.txt"), + Attr: Attr{ + Mode: 0o644, + Inode: 42, + }, + } + require.NoError(t, wrapper.InsertEntry(ctx, entry)) + require.NoError(t, wrapper.UpdateEntry(ctx, entry)) + assert.Zero(t, countInodeIndexRows(store)) + require.NoError(t, wrapper.DeleteEntry(ctx, entry.FullPath)) + assert.Zero(t, countInodeIndexRows(store)) +} + +func TestInodeIndexScopedToConfiguredPrefixes(t *testing.T) { + wrapper, store := newInodeIndexWrapper("/exports") + ctx := context.Background() + + require.NoError(t, wrapper.InsertEntry(ctx, &Entry{ + FullPath: util.FullPath("/exports/docs/in.txt"), + Attr: Attr{Mode: 0o644, Inode: 1001}, + })) + require.NoError(t, wrapper.InsertEntry(ctx, &Entry{ + FullPath: util.FullPath("/other/out.txt"), + Attr: Attr{Mode: 0o644, Inode: 1002}, + })) + + assert.Equal(t, 1, countInodeIndexRows(store)) + path, err := wrapper.lookupInodePath(ctx, 1001) + require.NoError(t, err) + assert.Equal(t, util.FullPath("/exports/docs/in.txt"), path) + _, err = wrapper.lookupInodePath(ctx, 1002) + require.ErrorIs(t, err, ErrKvNotFound) + + require.NoError(t, wrapper.InsertEntry(ctx, &Entry{ + FullPath: util.FullPath("/exports/docs"), + Attr: Attr{Mode: os.ModeDir | 0o755, Inode: 1000}, + })) + require.NoError(t, wrapper.DeleteFolderChildren(ctx, util.FullPath("/exports"))) + _, err = wrapper.lookupInodePath(ctx, 1001) + require.ErrorIs(t, err, ErrKvNotFound) +} + +func TestRecursiveDeleteRemovesDescendantInodeIndexes(t *testing.T) { + f, _ := newTestFilerWithStubStore() + fc := inodeIndexConf("/tree") + f.Store.(*FilerStoreWrapper).filerConfFn = func() *FilerConf { return fc } + ctx := context.Background() + + entries := []*Entry{ + { + FullPath: util.FullPath("/tree"), + Attr: Attr{ + Mode: os.ModeDir | 0o755, + Inode: 100, + }, + }, + { + FullPath: util.FullPath("/tree/file.txt"), + Attr: Attr{ + Mode: 0o644, + Inode: 101, + }, + }, + { + FullPath: util.FullPath("/tree/subdir"), + Attr: Attr{ + Mode: os.ModeDir | 0o755, + Inode: 102, + }, + }, + { + FullPath: util.FullPath("/tree/subdir/nested.txt"), + Attr: Attr{ + Mode: 0o644, + Inode: 103, + }, + }, + } + + for _, entry := range entries { + require.NoError(t, f.Store.InsertEntry(ctx, entry)) + } + + require.NoError(t, f.DeleteEntryMetaAndData(ctx, util.FullPath("/tree"), true, false, false, false, nil, 0)) + + for _, inode := range []uint64{100, 101, 102, 103} { + _, err := f.Store.(*FilerStoreWrapper).lookupInodePath(ctx, inode) + require.ErrorIs(t, err, ErrKvNotFound) + } +} + +func TestInodeIndexActiveUnder(t *testing.T) { + fc := inodeIndexConf("/exports") + assert.True(t, fc.InodeIndexActiveUnder("/")) + assert.True(t, fc.InodeIndexActiveUnder("/exports")) + assert.True(t, fc.InodeIndexActiveUnder("/exports/sub")) + assert.False(t, fc.InodeIndexActiveUnder("/other")) + + assert.False(t, NewFilerConf().InodeIndexActiveUnder("/")) +} diff --git a/weed/filer/filerstore_wrapper.go b/weed/filer/filerstore_wrapper.go index d742380ae..901baa88f 100644 --- a/weed/filer/filerstore_wrapper.go +++ b/weed/filer/filerstore_wrapper.go @@ -37,7 +37,8 @@ type FilerStoreWrapper struct { defaultStore FilerStore pathToStore ptrie.Trie[string] storeIdToStore map[string]FilerStore - hasPathSpecificStore bool // fast check to skip MatchPrefix when no path-specific stores + hasPathSpecificStore bool // fast check to skip MatchPrefix when no path-specific stores + filerConfFn func() *FilerConf // set by Filer.SetStore; nil disables the inode index } func NewFilerStoreWrapper(store FilerStore) *FilerStoreWrapper { @@ -151,7 +152,11 @@ func (fsw *FilerStoreWrapper) InsertEntry(ctx context.Context, entry *Entry) err return err } - return actualStore.InsertEntry(ctx, entry) + if err := actualStore.InsertEntry(ctx, entry); err != nil { + return err + } + fsw.recordInodeIndexWrite(ctx, "InsertEntry", entry.FullPath, entry.Attr.Inode) + return nil } // InsertEntryKnownAbsent skips the pre-insert FindEntry path when the caller has @@ -179,7 +184,11 @@ func (fsw *FilerStoreWrapper) InsertEntryKnownAbsent(ctx context.Context, entry } } - return actualStore.InsertEntry(ctx, entry) + if err := actualStore.InsertEntry(ctx, entry); err != nil { + return err + } + fsw.recordInodeIndexWrite(ctx, "InsertEntryKnownAbsent", entry.FullPath, entry.Attr.Inode) + return nil } func (fsw *FilerStoreWrapper) UpdateEntry(ctx context.Context, entry *Entry) error { @@ -206,7 +215,11 @@ func (fsw *FilerStoreWrapper) UpdateEntry(ctx context.Context, entry *Entry) err return err } - return actualStore.UpdateEntry(ctx, entry) + if err := actualStore.UpdateEntry(ctx, entry); err != nil { + return err + } + fsw.recordInodeIndexWrite(ctx, "UpdateEntry", entry.FullPath, entry.Attr.Inode) + return nil } func normalizeEntryMimeForStore(entry *Entry) { @@ -258,6 +271,8 @@ func (fsw *FilerStoreWrapper) DeleteEntry(ctx context.Context, fp util.FullPath) if findErr == filer_pb.ErrNotFound || existingEntry == nil { return nil } + inode := existingEntry.Attr.Inode + fullPath := existingEntry.FullPath if len(existingEntry.HardLinkId) != 0 { // remove hard link op := ctx.Value("OP") @@ -272,7 +287,11 @@ func (fsw *FilerStoreWrapper) DeleteEntry(ctx context.Context, fp util.FullPath) } } - return actualStore.DeleteEntry(ctx, fp) + if err := actualStore.DeleteEntry(ctx, fp); err != nil { + return err + } + fsw.recordInodeIndexRemoval(ctx, "DeleteEntry", fullPath, inode) + return nil } func (fsw *FilerStoreWrapper) DeleteOneEntry(ctx context.Context, existingEntry *Entry) (err error) { @@ -280,6 +299,8 @@ func (fsw *FilerStoreWrapper) DeleteOneEntry(ctx context.Context, existingEntry return err } ctx = context.WithoutCancel(ctx) + fullPath := existingEntry.FullPath + inode := existingEntry.Attr.Inode actualStore := fsw.getActualStore(existingEntry.FullPath) stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "delete").Inc() start := time.Now() @@ -302,7 +323,11 @@ func (fsw *FilerStoreWrapper) DeleteOneEntry(ctx context.Context, existingEntry } } - return actualStore.DeleteEntry(ctx, existingEntry.FullPath) + if err := actualStore.DeleteEntry(ctx, existingEntry.FullPath); err != nil { + return err + } + fsw.recordInodeIndexRemoval(ctx, "DeleteOneEntry", fullPath, inode) + return nil } func (fsw *FilerStoreWrapper) DeleteFolderChildren(ctx context.Context, fp util.FullPath) (err error) { @@ -317,7 +342,20 @@ func (fsw *FilerStoreWrapper) DeleteFolderChildren(ctx context.Context, fp util. stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "deleteFolderChildren").Observe(time.Since(start).Seconds()) }() - return actualStore.DeleteFolderChildren(ctx, fp) + collected, collectErr := fsw.collectInodeIndexEntries(ctx, fp) + if collectErr != nil { + // Index collection is best-effort: a failure here only prevents inode + // index housekeeping, not the directory removal itself. + glog.WarningfCtx(ctx, "collectInodeIndexEntries %s: %v; deleting folder children without index cleanup", fp, collectErr) + collected = nil + } + if err := actualStore.DeleteFolderChildren(ctx, fp); err != nil { + return err + } + for _, entry := range collected { + fsw.recordInodeIndexRemoval(ctx, "DeleteFolderChildren", entry.path, entry.inode) + } + return nil } func (fsw *FilerStoreWrapper) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc ListEachEntryFunc) (string, error) {