gitlab-com-seaweedfs/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-01-19 02:48:24 +00:00
Code Issues Projects Releases Wiki Activity

line wrap

Browse source
This commit is contained in:
Chris Lu 2021-08-31 23:30:28 -07:00
parent 68ea99190f
commit 8e125339d5
6 changed files with 22 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
weed/command/filer_backup.go
View file

@ -112,6 +112,7 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti
return setOffset(grpcDialOption, sourceFiler, BackupKeyPrefix, int32(sinkId), lastTsNs) return setOffset(grpcDialOption, sourceFiler, BackupKeyPrefix, int32(sinkId), lastTsNs)
}) })
return pb.FollowMetadata(sourceFiler, grpcDialOption, "backup_"+dataSink.GetName(), sourcePath, nil, startFrom.UnixNano(), 0, processEventFnWithOffset, false) return pb.FollowMetadata(sourceFiler, grpcDialOption, "backup_"+dataSink.GetName(),
sourcePath, nil, startFrom.UnixNano(), 0, processEventFnWithOffset, false)
} }

3
weed/command/filer_meta_backup.go
View file

@ -195,7 +195,8 @@ func (metaBackup *FilerMetaBackupOptions) streamMetadataBackup() error {
return metaBackup.setOffset(lastTime) return metaBackup.setOffset(lastTime)
}) })
return pb.FollowMetadata(*metaBackup.filerAddress, metaBackup.grpcDialOption, "meta_backup", *metaBackup.filerDirectory, nil, startTime.UnixNano(), 0, processEventFnWithOffset, false) return pb.FollowMetadata(*metaBackup.filerAddress, metaBackup.grpcDialOption, "meta_backup",
*metaBackup.filerDirectory, nil, startTime.UnixNano(), 0, processEventFnWithOffset, false)
} }

18
weed/command/filer_meta_tail.go
View file

@ -103,15 +103,17 @@ func runFilerMetaTail(cmd *Command, args []string) bool {
} }
} }
tailErr := pb.FollowMetadata(*tailFiler, grpcDialOption, "tail", *tailTarget, nil, time.Now().Add(-*tailStart).UnixNano(), 0, func(resp *filer_pb.SubscribeMetadataResponse) error { tailErr := pb.FollowMetadata(*tailFiler, grpcDialOption, "tail",
if !shouldPrint(resp) { *tailTarget, nil, time.Now().Add(-*tailStart).UnixNano(), 0,
func(resp *filer_pb.SubscribeMetadataResponse) error {
if !shouldPrint(resp) {
return nil
}
if err := eachEntryFunc(resp); err != nil {
return err
}
return nil return nil
} }, false)
if err := eachEntryFunc(resp); err != nil {
return err
}
return nil
}, false)
if tailErr != nil { if tailErr != nil {
fmt.Printf("tail %s: %v\n", *tailFiler, tailErr) fmt.Printf("tail %s: %v\n", *tailFiler, tailErr)

3
weed/command/filer_remote_sync.go
View file

@ -210,7 +210,8 @@ func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *sour
return setOffset(option.grpcDialOption, *option.filerAddress, RemoteSyncKeyPrefix, int32(dirHash), lastTsNs) return setOffset(option.grpcDialOption, *option.filerAddress, RemoteSyncKeyPrefix, int32(dirHash), lastTsNs)
}) })
return pb.FollowMetadata(*option.filerAddress, option.grpcDialOption, "filer.remote.sync", mountedDir, nil, lastOffsetTs.UnixNano(), 0, processEventFnWithOffset, false) return pb.FollowMetadata(*option.filerAddress, option.grpcDialOption, "filer.remote.sync",
mountedDir, nil, lastOffsetTs.UnixNano(), 0, processEventFnWithOffset, false)
} }
func toRemoteStorageLocation(mountDir, sourcePath util.FullPath, remoteMountLocation *remote_pb.RemoteStorageLocation) *remote_pb.RemoteStorageLocation { func toRemoteStorageLocation(mountDir, sourcePath util.FullPath, remoteMountLocation *remote_pb.RemoteStorageLocation) *remote_pb.RemoteStorageLocation {

3
weed/command/filer_sync.go
View file

@ -170,7 +170,8 @@ func doSubscribeFilerMetaChanges(grpcDialOption grpc.DialOption, sourceFiler, so
return setOffset(grpcDialOption, targetFiler, SyncKeyPrefix, sourceFilerSignature, lastTsNs) return setOffset(grpcDialOption, targetFiler, SyncKeyPrefix, sourceFilerSignature, lastTsNs)
}) })
return pb.FollowMetadata(sourceFiler, grpcDialOption, "syncTo_"+targetFiler, sourcePath, nil, sourceFilerOffsetTsNs, targetFilerSignature, processEventFnWithOffset, false) return pb.FollowMetadata(sourceFiler, grpcDialOption, "syncTo_"+targetFiler,
sourcePath, nil, sourceFilerOffsetTsNs, targetFilerSignature, processEventFnWithOffset, false)
} }

6
weed/pb/filer_pb_tail.go
View file

@ -28,7 +28,8 @@ func WithFilerClientFollowMetadata(filerClient filer_pb.FilerClient,
clientName string, pathPrefix string, lastTsNs int64, selfSignature int32, clientName string, pathPrefix string, lastTsNs int64, selfSignature int32,
processEventFn ProcessMetadataFunc, fatalOnError bool) error { processEventFn ProcessMetadataFunc, fatalOnError bool) error {
err := filerClient.WithFilerClient(makeFunc(clientName, pathPrefix, nil, lastTsNs, selfSignature, processEventFn, fatalOnError)) err := filerClient.WithFilerClient(makeFunc(clientName,
pathPrefix, nil, lastTsNs, selfSignature, processEventFn, fatalOnError))
if err != nil { if err != nil {
return fmt.Errorf("subscribing filer meta change: %v", err) return fmt.Errorf("subscribing filer meta change: %v", err)
} }
@ -36,7 +37,8 @@ func WithFilerClientFollowMetadata(filerClient filer_pb.FilerClient,
return nil return nil
} }
func makeFunc(clientName string, pathPrefix string, additionalPathPrefixes []string, lastTsNs int64, selfSignature int32, processEventFn ProcessMetadataFunc, fatalOnError bool) func(client filer_pb.SeaweedFilerClient) error { func makeFunc(clientName string, pathPrefix string, additionalPathPrefixes []string, lastTsNs int64, selfSignature int32,
processEventFn ProcessMetadataFunc, fatalOnError bool) func(client filer_pb.SeaweedFilerClient) error {
return func(client filer_pb.SeaweedFilerClient) error { return func(client filer_pb.SeaweedFilerClient) error {
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
defer cancel() defer cancel()