From 215b1695623a4ebcba5f45488a05d19857436a6f Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 1 Jul 2021 01:19:28 -0700 Subject: [PATCH] mount: recursively rename locally --- go.mod | 2 +- go.sum | 2 + other/java/client/src/main/proto/filer.proto | 1 + weed/filesys/dir_rename.go | 127 +++++++++++++++---- weed/filesys/meta_cache/meta_cache.go | 34 +++-- weed/pb/filer.proto | 1 + weed/pb/filer_pb/filer.pb.go | 22 +++- weed/server/filer_grpc_server_rename.go | 18 +-- weed/server/filer_grpc_server_sub_meta.go | 15 ++- 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)