mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/chrislusf/seaweedfs synced 2024-05-09 04:50:47 +02:00
Code Issues Projects Releases Wiki Activity

fix filer.sync missing source srv uploaded files to target when target down

Browse source
This commit is contained in:
creeew 2022-06-02 01:28:47 +08:00
parent 6793bc853c
commit 02ae102731
11 changed files with 36 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
unmaintained/load_test/load_test_meta_tail/load_test_meta_tail.go
View file

@ -77,7 +77,7 @@ func startGenerateMetadata() {
func startSubscribeMetadata(eachEntryFunc func(event *filer_pb.SubscribeMetadataResponse) error) { func startSubscribeMetadata(eachEntryFunc func(event *filer_pb.SubscribeMetadataResponse) error) {
tailErr := pb.FollowMetadata(pb.ServerAddress(*tailFiler), grpc.WithInsecure(), "tail", 0, *dir, nil, 0, 0, 0, eachEntryFunc, false) tailErr := pb.FollowMetadata(pb.ServerAddress(*tailFiler), grpc.WithInsecure(), "tail", 0, *dir, nil, 0, 0, 0, eachEntryFunc, pb.TrivialOnError)
if tailErr != nil { if tailErr != nil {
fmt.Printf("tail %s: %v\n", *tailFiler, tailErr) fmt.Printf("tail %s: %v\n", *tailFiler, tailErr)

2
weed/command/filer_backup.go
View file

@ -114,6 +114,6 @@ 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(), clientId, sourcePath, nil, startFrom.UnixNano(), 0, 0, processEventFnWithOffset, false) return pb.FollowMetadata(sourceFiler, grpcDialOption, "backup_"+dataSink.GetName(), clientId, sourcePath, nil, startFrom.UnixNano(), 0, 0, processEventFnWithOffset, pb.TrivialOnError)
} }

2
weed/command/filer_meta_backup.go
View file

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

2
weed/command/filer_meta_tail.go
View file

@ -119,7 +119,7 @@ func runFilerMetaTail(cmd *Command, args []string) bool {
return err return err
} }
return nil return nil
}, false) }, pb.TrivialOnError)
if tailErr != nil { if tailErr != nil {
fmt.Printf("tail %s: %v\n", *tailFiler, tailErr) fmt.Printf("tail %s: %v\n", *tailFiler, tailErr)

2
weed/command/filer_remote_gateway_buckets.go
View file

@ -39,7 +39,7 @@ func (option *RemoteGatewayOptions) followBucketUpdatesAndUploadToRemote(filerSo
lastOffsetTs := collectLastSyncOffset(option, option.grpcDialOption, pb.ServerAddress(*option.filerAddress), option.bucketsDir, *option.timeAgo) lastOffsetTs := collectLastSyncOffset(option, option.grpcDialOption, pb.ServerAddress(*option.filerAddress), option.bucketsDir, *option.timeAgo)
return pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, "filer.remote.sync", option.clientId, return pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, "filer.remote.sync", option.clientId,
option.bucketsDir, []string{filer.DirectoryEtcRemote}, lastOffsetTs.UnixNano(), 0, 0, processEventFnWithOffset, false) option.bucketsDir, []string{filer.DirectoryEtcRemote}, lastOffsetTs.UnixNano(), 0, 0, processEventFnWithOffset, pb.TrivialOnError)
} }
func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *source.FilerSource) (pb.ProcessMetadataFunc, error) { func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *source.FilerSource) (pb.ProcessMetadataFunc, error) {

2
weed/command/filer_remote_sync_dir.go
View file

@ -41,7 +41,7 @@ func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *sour
lastOffsetTs := collectLastSyncOffset(option, option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir, *option.timeAgo) lastOffsetTs := collectLastSyncOffset(option, option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir, *option.timeAgo)
return pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, "filer.remote.sync", option.clientId, return pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, "filer.remote.sync", option.clientId,
mountedDir, []string{filer.DirectoryEtcRemote}, lastOffsetTs.UnixNano(), 0, 0, processEventFnWithOffset, false) mountedDir, []string{filer.DirectoryEtcRemote}, lastOffsetTs.UnixNano(), 0, 0, processEventFnWithOffset, pb.TrivialOnError)
} }
func makeEventProcessor(remoteStorage *remote_pb.RemoteConf, mountedDir string, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, filerSource *source.FilerSource) (pb.ProcessMetadataFunc, error) { func makeEventProcessor(remoteStorage *remote_pb.RemoteConf, mountedDir string, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, filerSource *source.FilerSource) (pb.ProcessMetadataFunc, error) {

2
weed/command/filer_sync.go
View file

@ -178,7 +178,7 @@ func doSubscribeFilerMetaChanges(clientId int32, grpcDialOption grpc.DialOption,
}) })
return pb.FollowMetadata(sourceFiler, grpcDialOption, "syncTo_"+string(targetFiler), clientId, return pb.FollowMetadata(sourceFiler, grpcDialOption, "syncTo_"+string(targetFiler), clientId,
sourcePath, nil, sourceFilerOffsetTsNs, 0, targetFilerSignature, processEventFnWithOffset, false) sourcePath, nil, sourceFilerOffsetTsNs, 0, targetFilerSignature, processEventFnWithOffset, pb.RetryForeverOnError)
} }

2
weed/filer/filer.go
View file

@ -85,7 +85,7 @@ func (f *Filer) MaybeBootstrapFromPeers(self pb.ServerAddress, existingNodes []*
err = pb.FollowMetadata(pb.ServerAddress(earliestNode.Address), f.GrpcDialOption, "bootstrap", int32(f.UniqueFileId), "/", nil, err = pb.FollowMetadata(pb.ServerAddress(earliestNode.Address), f.GrpcDialOption, "bootstrap", int32(f.UniqueFileId), "/", nil,
0, snapshotTime.UnixNano(), f.Signature, func(resp *filer_pb.SubscribeMetadataResponse) error { 0, snapshotTime.UnixNano(), f.Signature, func(resp *filer_pb.SubscribeMetadataResponse) error {
return Replay(f.Store, resp) return Replay(f.Store, resp)
}, true) }, pb.FatalOnError)
return return
} }

2
weed/mount/meta_cache/meta_cache_subscribe.go
View file

@ -58,7 +58,7 @@ func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.Fil
} }
util.RetryForever("followMetaUpdates", func() error { util.RetryForever("followMetaUpdates", func() error {
return pb.WithFilerClientFollowMetadata(client, "mount", selfSignature, dir, &lastTsNs, 0, selfSignature, processEventFn, true) return pb.WithFilerClientFollowMetadata(client, "mount", selfSignature, dir, &lastTsNs, 0, selfSignature, processEventFn, pb.FatalOnError)
}, func(err error) bool { }, func(err error) bool {
glog.Errorf("follow metadata updates: %v", err) glog.Errorf("follow metadata updates: %v", err)
return true return true

33
weed/pb/filer_pb_tail.go
View file

@ -5,19 +5,28 @@ import (
"fmt" "fmt"
"github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb" "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/util"
"google.golang.org/grpc" "google.golang.org/grpc"
"io" "io"
"time" "time"
) )
type EventErrorType int
const (
TrivialOnError EventErrorType = iota
FatalOnError
RetryForeverOnError
)
type ProcessMetadataFunc func(resp *filer_pb.SubscribeMetadataResponse) error type ProcessMetadataFunc func(resp *filer_pb.SubscribeMetadataResponse) error
func FollowMetadata(filerAddress ServerAddress, grpcDialOption grpc.DialOption, clientName string, clientId int32, func FollowMetadata(filerAddress ServerAddress, grpcDialOption grpc.DialOption, clientName string, clientId int32,
pathPrefix string, additionalPathPrefixes []string, lastTsNs int64, untilTsNs int64, selfSignature int32, pathPrefix string, additionalPathPrefixes []string, lastTsNs int64, untilTsNs int64, selfSignature int32,
processEventFn ProcessMetadataFunc, fatalOnError bool) error { processEventFn ProcessMetadataFunc, eventErrorType EventErrorType) error {
err := WithFilerClient(true, filerAddress, grpcDialOption, makeSubscribeMetadataFunc(clientName, clientId, err := WithFilerClient(true, filerAddress, grpcDialOption, makeSubscribeMetadataFunc(clientName, clientId,
pathPrefix, additionalPathPrefixes, &lastTsNs, untilTsNs, selfSignature, processEventFn, fatalOnError)) pathPrefix, additionalPathPrefixes, &lastTsNs, untilTsNs, selfSignature, processEventFn, eventErrorType))
if err != nil { if err != nil {
return fmt.Errorf("subscribing filer meta change: %v", err) return fmt.Errorf("subscribing filer meta change: %v", err)
} }
@ -26,10 +35,10 @@ func FollowMetadata(filerAddress ServerAddress, grpcDialOption grpc.DialOption,
func WithFilerClientFollowMetadata(filerClient filer_pb.FilerClient, func WithFilerClientFollowMetadata(filerClient filer_pb.FilerClient,
clientName string, clientId int32, pathPrefix string, lastTsNs *int64, untilTsNs int64, selfSignature int32, clientName string, clientId int32, pathPrefix string, lastTsNs *int64, untilTsNs int64, selfSignature int32,
processEventFn ProcessMetadataFunc, fatalOnError bool) error { processEventFn ProcessMetadataFunc, eventErrorType EventErrorType) error {
err := filerClient.WithFilerClient(true, makeSubscribeMetadataFunc(clientName, clientId, err := filerClient.WithFilerClient(true, makeSubscribeMetadataFunc(clientName, clientId,
pathPrefix, nil, lastTsNs, untilTsNs, selfSignature, processEventFn, fatalOnError)) pathPrefix, nil, lastTsNs, untilTsNs, selfSignature, processEventFn, eventErrorType))
if err != nil { if err != nil {
return fmt.Errorf("subscribing filer meta change: %v", err) return fmt.Errorf("subscribing filer meta change: %v", err)
} }
@ -38,7 +47,7 @@ func WithFilerClientFollowMetadata(filerClient filer_pb.FilerClient,
} }
func makeSubscribeMetadataFunc(clientName string, clientId int32, pathPrefix string, additionalPathPrefixes []string, lastTsNs *int64, untilTsNs int64, selfSignature int32, func makeSubscribeMetadataFunc(clientName string, clientId int32, pathPrefix string, additionalPathPrefixes []string, lastTsNs *int64, untilTsNs int64, selfSignature int32,
processEventFn ProcessMetadataFunc, fatalOnError bool) func(client filer_pb.SeaweedFilerClient) error { processEventFn ProcessMetadataFunc, eventErrorType EventErrorType) 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()
@ -65,9 +74,19 @@ func makeSubscribeMetadataFunc(clientName string, clientId int32, pathPrefix str
} }
if err := processEventFn(resp); err != nil { if err := processEventFn(resp); err != nil {
if fatalOnError { switch eventErrorType {
case TrivialOnError:
glog.Errorf("process %v: %v", resp, err)
case FatalOnError:
glog.Fatalf("process %v: %v", resp, err) glog.Fatalf("process %v: %v", resp, err)
} else { case RetryForeverOnError:
util.RetryForever("followMetaUpdates", func() error {
return processEventFn(resp)
}, func(err error) bool {
glog.Errorf("process %v: %v", resp, err)
return true
})
default:
glog.Errorf("process %v: %v", resp, err) glog.Errorf("process %v: %v", resp, err)
} }
} }

2
weed/s3api/auth_credentials_subscribe.go
View file

@ -33,7 +33,7 @@ func (s3a *S3ApiServer) subscribeMetaEvents(clientName string, prefix string, la
} }
util.RetryForever("followIamChanges", func() error { util.RetryForever("followIamChanges", func() error {
return pb.WithFilerClientFollowMetadata(s3a, clientName, s3a.randomClientId, prefix, &lastTsNs, 0, 0, processEventFn, true) return pb.WithFilerClientFollowMetadata(s3a, clientName, s3a.randomClientId, prefix, &lastTsNs, 0, 0, processEventFn, pb.FatalOnError)
}, func(err error) bool { }, func(err error) bool {
glog.V(0).Infof("iam follow metadata changes: %v", err) glog.V(0).Infof("iam follow metadata changes: %v", err)
return true return true