aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--go.mod2
-rw-r--r--go.sum2
-rw-r--r--other/java/client/src/main/proto/filer.proto1
-rw-r--r--weed/filesys/dir_rename.go127
-rw-r--r--weed/filesys/meta_cache/meta_cache.go34
-rw-r--r--weed/pb/filer.proto1
-rw-r--r--weed/pb/filer_pb/filer.pb.go22
-rw-r--r--weed/server/filer_grpc_server_rename.go18
-rw-r--r--weed/server/filer_grpc_server_sub_meta.go15
9 files changed, 162 insertions, 60 deletions
diff --git a/go.mod b/go.mod
index 5510eee8e..84c03ebdc 100644
--- a/go.mod
+++ b/go.mod
@@ -56,7 +56,7 @@ require (
github.com/pquerna/cachecontrol v0.1.0
github.com/prometheus/client_golang v1.11.0
github.com/rcrowley/go-metrics v0.0.0-20190826022208-cac0b30c2563 // indirect
- github.com/seaweedfs/fuse v1.1.7
+ github.com/seaweedfs/fuse v1.1.8
github.com/seaweedfs/goexif v1.0.2
github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e
github.com/spaolacci/murmur3 v1.1.0 // indirect
diff --git a/go.sum b/go.sum
index e0587fd9c..1911be8dd 100644
--- a/go.sum
+++ b/go.sum
@@ -617,6 +617,8 @@ github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da/go.mod h1:gi+0
github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
github.com/seaweedfs/fuse v1.1.7 h1:T4L5c/Sn+q8lE+0zCmH2MWvIO+B5TttWOSqK5KQPRMQ=
github.com/seaweedfs/fuse v1.1.7/go.mod h1:+PP6WlkrRUG6KPE+Th2EX5To/PjHaFsvqg/UgQ39aj8=
+github.com/seaweedfs/fuse v1.1.8 h1:YFSDPotG4uhQzV7ooDUvQ8BRVy5rM1XCFPJAmAsZz68=
+github.com/seaweedfs/fuse v1.1.8/go.mod h1:+PP6WlkrRUG6KPE+Th2EX5To/PjHaFsvqg/UgQ39aj8=
github.com/seaweedfs/goexif v1.0.2 h1:p+rTXYdQ2mgxd+1JaTrQ9N8DvYuw9UH9xgYmJ+Bb29E=
github.com/seaweedfs/goexif v1.0.2/go.mod h1:MrKs5LK0HXdffrdCZrW3OIMegL2xXpC6ThLyXMyjdrk=
github.com/secsy/goftp v0.0.0-20190720192957-f31499d7c79a h1:C6IhVTxNkhlb0tlCB6JfHOUv1f0xHPK7V8X4HlJZEJw=
diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto
index cdbba7eb1..f49c1218f 100644
--- a/other/java/client/src/main/proto/filer.proto
+++ b/other/java/client/src/main/proto/filer.proto
@@ -208,6 +208,7 @@ message AtomicRenameEntryRequest {
string old_name = 2;
string new_directory = 3;
string new_name = 4;
+ repeated int32 signatures = 5;
}
message AtomicRenameEntryResponse {
diff --git a/weed/filesys/dir_rename.go b/weed/filesys/dir_rename.go
index d50c6dab0..27a14d5c6 100644
--- a/weed/filesys/dir_rename.go
+++ b/weed/filesys/dir_rename.go
@@ -2,6 +2,9 @@ package filesys
import (
"context"
+ "fmt"
+ "github.com/chrislusf/seaweedfs/weed/filer"
+ "math"
"github.com/seaweedfs/fuse"
"github.com/seaweedfs/fuse/fs"
@@ -37,6 +40,7 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
OldName: req.OldName,
NewDirectory: newDir.FullPath(),
NewName: req.NewName,
+ Signatures: []int32{dir.wfs.signature},
}
_, err := client.AtomicRenameEntry(ctx, request)
@@ -53,34 +57,12 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
return fuse.EIO
}
- // TODO: replicate renaming logic on filer
- if err := dir.wfs.metaCache.DeleteEntry(context.Background(), oldPath); err != nil {
- glog.V(0).Infof("dir Rename delete local %s => %s : %v", oldPath, newPath, err)
- return fuse.EIO
- }
- oldEntry.FullPath = newPath
- if err := dir.wfs.metaCache.InsertEntry(context.Background(), oldEntry); err != nil {
- glog.V(0).Infof("dir Rename insert local %s => %s : %v", oldPath, newPath, err)
+ err = dir.moveEntry(context.Background(), util.FullPath(dir.FullPath()), oldEntry, util.FullPath(newDir.FullPath()), req.NewName)
+ if err != nil {
+ glog.V(0).Infof("dir local Rename %s => %s : %v", oldPath, newPath, err)
return fuse.EIO
}
- oldFsNode := NodeWithId(oldPath.AsInode())
- newFsNode := NodeWithId(newPath.AsInode())
- dir.wfs.Server.InvalidateInternalNode(oldFsNode, newFsNode, func(internalNode fs.Node) {
- if file, ok := internalNode.(*File); ok {
- glog.V(4).Infof("internal file node %s", file.Name)
- file.Name = req.NewName
- file.id = uint64(newFsNode)
- file.dir = newDir
- }
- if dir, ok := internalNode.(*Dir); ok {
- glog.V(4).Infof("internal dir node %s", dir.name)
- dir.name = req.NewName
- dir.id = uint64(newFsNode)
- dir.parent = newDir
- }
- })
-
// change file handle
dir.wfs.handlesLock.Lock()
defer dir.wfs.handlesLock.Unlock()
@@ -96,3 +78,98 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
return nil
}
+
+func (dir *Dir) moveEntry(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string) error {
+
+ oldName := entry.Name()
+
+ if err := dir.moveSelfEntry(ctx, oldParent, entry, newParent, newName, func() error {
+
+ oldFsNode := NodeWithId(oldParent.Child(oldName).AsInode())
+ newFsNode := NodeWithId(newParent.Child(newName).AsInode())
+ newDirNode, found := dir.wfs.Server.FindInternalNode(NodeWithId(newParent.AsInode()))
+ var newDir *Dir
+ if found {
+ newDir = newDirNode.(*Dir)
+ }
+ dir.wfs.Server.InvalidateInternalNode(oldFsNode, newFsNode, func(internalNode fs.Node) {
+ if file, ok := internalNode.(*File); ok {
+ glog.V(4).Infof("internal file node %s", oldParent.Child(oldName))
+ file.Name = newName
+ file.id = uint64(newFsNode)
+ if found {
+ file.dir = newDir
+ }
+ }
+ if dir, ok := internalNode.(*Dir); ok {
+ glog.V(4).Infof("internal dir node %s", oldParent.Child(oldName))
+ dir.name = newName
+ dir.id = uint64(newFsNode)
+ if found {
+ dir.parent = newDir
+ }
+ }
+ })
+
+ if entry.IsDirectory() {
+ if err := dir.moveFolderSubEntries(ctx, oldParent, oldName, newParent, newName); err != nil {
+ return err
+ }
+ }
+ return nil
+ }); err != nil {
+ return fmt.Errorf("fail to move %s => %s: %v", oldParent.Child(oldName), newParent.Child(newName), err)
+ }
+
+ return nil
+}
+
+func (dir *Dir) moveFolderSubEntries(ctx context.Context, oldParent util.FullPath, oldName string, newParent util.FullPath, newName string) error {
+
+ currentDirPath := oldParent.Child(oldName)
+ newDirPath := newParent.Child(newName)
+
+ glog.V(1).Infof("moving folder %s => %s", currentDirPath, newDirPath)
+
+ var moveErr error
+ listErr := dir.wfs.metaCache.ListDirectoryEntries(ctx, currentDirPath, "", false, int64(math.MaxInt32), func(item *filer.Entry) bool {
+ moveErr = dir.moveEntry(ctx, currentDirPath, item, newDirPath, item.Name())
+ if moveErr != nil {
+ return false
+ }
+ return true
+ })
+ if listErr != nil {
+ return listErr
+ }
+ if moveErr != nil {
+ return moveErr
+ }
+
+ return nil
+}
+
+func (dir *Dir) moveSelfEntry(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, moveFolderSubEntries func() error) error {
+
+ newPath := newParent.Child(newName)
+ oldPath := oldParent.Child(entry.Name())
+
+ entry.FullPath = newPath
+ if err := dir.wfs.metaCache.InsertEntry(ctx, entry); err != nil {
+ glog.V(0).Infof("dir Rename insert local %s => %s : %v", oldPath, newPath, err)
+ return fuse.EIO
+ }
+
+ if moveFolderSubEntries != nil {
+ if moveChildrenErr := moveFolderSubEntries(); moveChildrenErr != nil {
+ return moveChildrenErr
+ }
+ }
+
+ if err := dir.wfs.metaCache.DeleteEntry(ctx, oldPath); err != nil {
+ glog.V(0).Infof("dir Rename delete local %s => %s : %v", oldPath, newPath, err)
+ return fuse.EIO
+ }
+
+ return nil
+}
diff --git a/weed/filesys/meta_cache/meta_cache.go b/weed/filesys/meta_cache/meta_cache.go
index 3f6391c39..69d1655ee 100644
--- a/weed/filesys/meta_cache/meta_cache.go
+++ b/weed/filesys/meta_cache/meta_cache.go
@@ -3,14 +3,12 @@ package meta_cache
import (
"context"
"fmt"
- "os"
- "sync"
-
"github.com/chrislusf/seaweedfs/weed/filer"
"github.com/chrislusf/seaweedfs/weed/filer/leveldb"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/util"
"github.com/chrislusf/seaweedfs/weed/util/bounded_tree"
+ "os"
)
// need to have logic similar to FilerStoreWrapper
@@ -18,7 +16,7 @@ import (
type MetaCache struct {
localStore filer.VirtualFilerStore
- sync.RWMutex
+ // sync.RWMutex
visitedBoundary *bounded_tree.BoundedTree
uidGidMapper *UidGidMapper
invalidateFunc func(util.FullPath)
@@ -54,8 +52,8 @@ func openMetaStore(dbFolder string) filer.VirtualFilerStore {
}
func (mc *MetaCache) InsertEntry(ctx context.Context, entry *filer.Entry) error {
- mc.Lock()
- defer mc.Unlock()
+ //mc.Lock()
+ //defer mc.Unlock()
return mc.doInsertEntry(ctx, entry)
}
@@ -64,8 +62,8 @@ func (mc *MetaCache) doInsertEntry(ctx context.Context, entry *filer.Entry) erro
}
func (mc *MetaCache) AtomicUpdateEntryFromFiler(ctx context.Context, oldPath util.FullPath, newEntry *filer.Entry) error {
- mc.Lock()
- defer mc.Unlock()
+ //mc.Lock()
+ //defer mc.Unlock()
oldDir, _ := oldPath.DirAndName()
if mc.visitedBoundary.HasVisited(util.FullPath(oldDir)) {
@@ -97,14 +95,14 @@ func (mc *MetaCache) AtomicUpdateEntryFromFiler(ctx context.Context, oldPath uti
}
func (mc *MetaCache) UpdateEntry(ctx context.Context, entry *filer.Entry) error {
- mc.Lock()
- defer mc.Unlock()
+ //mc.Lock()
+ //defer mc.Unlock()
return mc.localStore.UpdateEntry(ctx, entry)
}
func (mc *MetaCache) FindEntry(ctx context.Context, fp util.FullPath) (entry *filer.Entry, err error) {
- mc.RLock()
- defer mc.RUnlock()
+ //mc.RLock()
+ //defer mc.RUnlock()
entry, err = mc.localStore.FindEntry(ctx, fp)
if err != nil {
return nil, err
@@ -114,14 +112,14 @@ func (mc *MetaCache) FindEntry(ctx context.Context, fp util.FullPath) (entry *fi
}
func (mc *MetaCache) DeleteEntry(ctx context.Context, fp util.FullPath) (err error) {
- mc.Lock()
- defer mc.Unlock()
+ //mc.Lock()
+ //defer mc.Unlock()
return mc.localStore.DeleteEntry(ctx, fp)
}
func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) error {
- mc.RLock()
- defer mc.RUnlock()
+ //mc.RLock()
+ //defer mc.RUnlock()
if !mc.visitedBoundary.HasVisited(dirPath) {
return fmt.Errorf("unsynchronized dir: %v", dirPath)
@@ -138,8 +136,8 @@ func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.Full
}
func (mc *MetaCache) Shutdown() {
- mc.Lock()
- defer mc.Unlock()
+ //mc.Lock()
+ //defer mc.Unlock()
mc.localStore.Shutdown()
}
diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto
index cdbba7eb1..f49c1218f 100644
--- a/weed/pb/filer.proto
+++ b/weed/pb/filer.proto
@@ -208,6 +208,7 @@ message AtomicRenameEntryRequest {
string old_name = 2;
string new_directory = 3;
string new_name = 4;
+ repeated int32 signatures = 5;
}
message AtomicRenameEntryResponse {
diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go
index 89fc448f4..2a7f3d041 100644
--- a/weed/pb/filer_pb/filer.pb.go
+++ b/weed/pb/filer_pb/filer.pb.go
@@ -1382,10 +1382,11 @@ type AtomicRenameEntryRequest struct {
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
- OldDirectory string `protobuf:"bytes,1,opt,name=old_directory,json=oldDirectory,proto3" json:"old_directory,omitempty"`
- OldName string `protobuf:"bytes,2,opt,name=old_name,json=oldName,proto3" json:"old_name,omitempty"`
- NewDirectory string `protobuf:"bytes,3,opt,name=new_directory,json=newDirectory,proto3" json:"new_directory,omitempty"`
- NewName string `protobuf:"bytes,4,opt,name=new_name,json=newName,proto3" json:"new_name,omitempty"`
+ OldDirectory string `protobuf:"bytes,1,opt,name=old_directory,json=oldDirectory,proto3" json:"old_directory,omitempty"`
+ OldName string `protobuf:"bytes,2,opt,name=old_name,json=oldName,proto3" json:"old_name,omitempty"`
+ NewDirectory string `protobuf:"bytes,3,opt,name=new_directory,json=newDirectory,proto3" json:"new_directory,omitempty"`
+ NewName string `protobuf:"bytes,4,opt,name=new_name,json=newName,proto3" json:"new_name,omitempty"`
+ Signatures []int32 `protobuf:"varint,5,rep,packed,name=signatures,proto3" json:"signatures,omitempty"`
}
func (x *AtomicRenameEntryRequest) Reset() {
@@ -1448,6 +1449,13 @@ func (x *AtomicRenameEntryRequest) GetNewName() string {
return ""
}
+func (x *AtomicRenameEntryRequest) GetSignatures() []int32 {
+ if x != nil {
+ return x.Signatures
+ }
+ return nil
+}
+
type AtomicRenameEntryResponse struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
@@ -3418,7 +3426,7 @@ var file_filer_proto_rawDesc = []byte{
0x72, 0x65, 0x73, 0x22, 0x2b, 0x0a, 0x13, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x45, 0x6e, 0x74,
0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x65, 0x72,
0x72, 0x6f, 0x72, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72,
- 0x22, 0x9a, 0x01, 0x0a, 0x18, 0x41, 0x74, 0x6f, 0x6d, 0x69, 0x63, 0x52, 0x65, 0x6e, 0x61, 0x6d,
+ 0x22, 0xba, 0x01, 0x0a, 0x18, 0x41, 0x74, 0x6f, 0x6d, 0x69, 0x63, 0x52, 0x65, 0x6e, 0x61, 0x6d,
0x65, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x23, 0x0a,
0x0d, 0x6f, 0x6c, 0x64, 0x5f, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x6f, 0x72, 0x79, 0x18, 0x01,
0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x6f, 0x6c, 0x64, 0x44, 0x69, 0x72, 0x65, 0x63, 0x74, 0x6f,
@@ -3427,7 +3435,9 @@ var file_filer_proto_rawDesc = []byte{
0x0d, 0x6e, 0x65, 0x77, 0x5f, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x6f, 0x72, 0x79, 0x18, 0x03,
0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x6e, 0x65, 0x77, 0x44, 0x69, 0x72, 0x65, 0x63, 0x74, 0x6f,
0x72, 0x79, 0x12, 0x19, 0x0a, 0x08, 0x6e, 0x65, 0x77, 0x5f, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x04,
- 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x6e, 0x65, 0x77, 0x4e, 0x61, 0x6d, 0x65, 0x22, 0x1b, 0x0a,
+ 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x6e, 0x65, 0x77, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x1e, 0x0a,
+ 0x0a, 0x73, 0x69, 0x67, 0x6e, 0x61, 0x74, 0x75, 0x72, 0x65, 0x73, 0x18, 0x05, 0x20, 0x03, 0x28,
+ 0x05, 0x52, 0x0a, 0x73, 0x69, 0x67, 0x6e, 0x61, 0x74, 0x75, 0x72, 0x65, 0x73, 0x22, 0x1b, 0x0a,
0x19, 0x41, 0x74, 0x6f, 0x6d, 0x69, 0x63, 0x52, 0x65, 0x6e, 0x61, 0x6d, 0x65, 0x45, 0x6e, 0x74,
0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0xec, 0x01, 0x0a, 0x13, 0x41,
0x73, 0x73, 0x69, 0x67, 0x6e, 0x56, 0x6f, 0x6c, 0x75, 0x6d, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65,
diff --git a/weed/server/filer_grpc_server_rename.go b/weed/server/filer_grpc_server_rename.go
index ba9f15370..8a11c91e3 100644
--- a/weed/server/filer_grpc_server_rename.go
+++ b/weed/server/filer_grpc_server_rename.go
@@ -33,7 +33,7 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom
return nil, fmt.Errorf("%s/%s not found: %v", req.OldDirectory, req.OldName, err)
}
- moveErr := fs.moveEntry(ctx, oldParent, oldEntry, newParent, req.NewName)
+ moveErr := fs.moveEntry(ctx, oldParent, oldEntry, newParent, req.NewName, req.Signatures)
if moveErr != nil {
fs.filer.RollbackTransaction(ctx)
return nil, fmt.Errorf("%s/%s move error: %v", req.OldDirectory, req.OldName, moveErr)
@@ -47,23 +47,23 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom
return &filer_pb.AtomicRenameEntryResponse{}, nil
}
-func (fs *FilerServer) moveEntry(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string) error {
+func (fs *FilerServer) moveEntry(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32) error {
if err := fs.moveSelfEntry(ctx, oldParent, entry, newParent, newName, func() error {
if entry.IsDirectory() {
- if err := fs.moveFolderSubEntries(ctx, oldParent, entry, newParent, newName); err != nil {
+ if err := fs.moveFolderSubEntries(ctx, oldParent, entry, newParent, newName, signatures); err != nil {
return err
}
}
return nil
- }); err != nil {
+ }, signatures); err != nil {
return fmt.Errorf("fail to move %s => %s: %v", oldParent.Child(entry.Name()), newParent.Child(newName), err)
}
return nil
}
-func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string) error {
+func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32) error {
currentDirPath := oldParent.Child(entry.Name())
newDirPath := newParent.Child(newName)
@@ -84,7 +84,7 @@ func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, oldParent util.
for _, item := range entries {
lastFileName = item.Name()
// println("processing", lastFileName)
- err := fs.moveEntry(ctx, currentDirPath, item, newDirPath, item.Name())
+ err := fs.moveEntry(ctx, currentDirPath, item, newDirPath, item.Name(), signatures)
if err != nil {
return err
}
@@ -96,7 +96,7 @@ func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, oldParent util.
return nil
}
-func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, moveFolderSubEntries func() error) error {
+func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, moveFolderSubEntries func() error, signatures []int32) error {
oldPath, newPath := oldParent.Child(entry.Name()), newParent.Child(newName)
@@ -115,7 +115,7 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent util.FullPat
Extended: entry.Extended,
Content: entry.Content,
}
- if createErr := fs.filer.CreateEntry(ctx, newEntry, false, false, nil); createErr != nil {
+ if createErr := fs.filer.CreateEntry(ctx, newEntry, false, false, signatures); createErr != nil {
return createErr
}
@@ -126,7 +126,7 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent util.FullPat
}
// delete old entry
- deleteErr := fs.filer.DeleteEntryMetaAndData(ctx, oldPath, false, false, false, false, nil)
+ deleteErr := fs.filer.DeleteEntryMetaAndData(ctx, oldPath, false, false, false, false, signatures)
if deleteErr != nil {
return deleteErr
}
diff --git a/weed/server/filer_grpc_server_sub_meta.go b/weed/server/filer_grpc_server_sub_meta.go
index 18505a95f..e79406bbe 100644
--- a/weed/server/filer_grpc_server_sub_meta.go
+++ b/weed/server/filer_grpc_server_sub_meta.go
@@ -25,7 +25,20 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
lastReadTime := time.Unix(0, req.SinceNs)
glog.V(0).Infof(" %v starts to subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
- eachEventNotificationFn := fs.eachEventNotificationFn(req, stream, clientName, req.Signature)
+ t := fs.eachEventNotificationFn(req, stream, clientName, req.Signature)
+
+ eachEventNotificationFn := func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error {
+ found := false
+ for _, sig := range eventNotification.Signatures {
+ if req.Signature == sig {
+ found = true
+ }
+ }
+ if !found {
+ glog.V(0).Infof("fresh message for %s(%d) %s %s", clientName, req.Signature, dirPath, eventNotification.String())
+ }
+ return t(dirPath, eventNotification, tsNs)
+ }
eachLogEntryFn := eachLogEntryFn(eachEventNotificationFn)
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-21 19:47:36 +0000