mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2024-01-19 02:48:24 +00:00
s3 to watch specific directories
This commit is contained in:
parent
b90d92fca9
commit
31922b2bf2
|
@ -60,7 +60,7 @@ func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.Fil
|
||||||
var clientEpoch int32
|
var clientEpoch int32
|
||||||
util.RetryForever("followMetaUpdates", func() error {
|
util.RetryForever("followMetaUpdates", func() error {
|
||||||
clientEpoch++
|
clientEpoch++
|
||||||
return pb.WithFilerClientFollowMetadata(client, "mount", selfSignature, clientEpoch, dir, &lastTsNs, 0, selfSignature, processEventFn, pb.FatalOnError)
|
return pb.WithFilerClientFollowMetadata(client, "mount", selfSignature, clientEpoch, dir, nil, &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
|
||||||
|
|
|
@ -25,20 +25,16 @@ func FollowMetadata(filerAddress ServerAddress, grpcDialOption grpc.DialOption,
|
||||||
pathPrefix string, additionalPathPrefixes []string, lastTsNs int64, untilTsNs int64, selfSignature int32,
|
pathPrefix string, additionalPathPrefixes []string, lastTsNs int64, untilTsNs int64, selfSignature int32,
|
||||||
processEventFn ProcessMetadataFunc, eventErrorType EventErrorType) error {
|
processEventFn ProcessMetadataFunc, eventErrorType EventErrorType) error {
|
||||||
|
|
||||||
err := WithFilerClient(true, filerAddress, grpcDialOption, makeSubscribeMetadataFunc(
|
err := WithFilerClient(true, filerAddress, grpcDialOption, makeSubscribeMetadataFunc(clientName, clientId, clientEpoch, pathPrefix, additionalPathPrefixes, nil, &lastTsNs, untilTsNs, selfSignature, processEventFn, eventErrorType))
|
||||||
clientName, clientId, clientEpoch,
|
|
||||||
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)
|
||||||
}
|
}
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func WithFilerClientFollowMetadata(filerClient filer_pb.FilerClient,
|
func WithFilerClientFollowMetadata(filerClient filer_pb.FilerClient, clientName string, clientId int32, clientEpoch int32, pathPrefix string, directoriesToWatch []string, lastTsNs *int64, untilTsNs int64, selfSignature int32, processEventFn ProcessMetadataFunc, eventErrorType EventErrorType) error {
|
||||||
clientName string, clientId int32, clientEpoch int32, pathPrefix string, lastTsNs *int64, untilTsNs int64, selfSignature int32,
|
|
||||||
processEventFn ProcessMetadataFunc, eventErrorType EventErrorType) error {
|
|
||||||
|
|
||||||
err := filerClient.WithFilerClient(true, makeSubscribeMetadataFunc(clientName, clientId, clientEpoch, pathPrefix, nil, lastTsNs, untilTsNs, selfSignature, processEventFn, eventErrorType))
|
err := filerClient.WithFilerClient(true, makeSubscribeMetadataFunc(clientName, clientId, clientEpoch, pathPrefix, nil, directoriesToWatch, 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)
|
||||||
}
|
}
|
||||||
|
@ -46,7 +42,7 @@ func WithFilerClientFollowMetadata(filerClient filer_pb.FilerClient,
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func makeSubscribeMetadataFunc(clientName string, clientId int32, clientEpoch int32, pathPrefix string, additionalPathPrefixes []string, lastTsNs *int64, untilTsNs int64, selfSignature int32, processEventFn ProcessMetadataFunc, eventErrorType EventErrorType) func(client filer_pb.SeaweedFilerClient) error {
|
func makeSubscribeMetadataFunc(clientName string, clientId int32, clientEpoch int32, pathPrefix string, additionalPathPrefixes []string, directoriesToWatch []string, lastTsNs *int64, untilTsNs int64, selfSignature int32, 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()
|
||||||
|
@ -54,6 +50,7 @@ func makeSubscribeMetadataFunc(clientName string, clientId int32, clientEpoch in
|
||||||
ClientName: clientName,
|
ClientName: clientName,
|
||||||
PathPrefix: pathPrefix,
|
PathPrefix: pathPrefix,
|
||||||
PathPrefixes: additionalPathPrefixes,
|
PathPrefixes: additionalPathPrefixes,
|
||||||
|
Directories: directoriesToWatch,
|
||||||
SinceNs: *lastTsNs,
|
SinceNs: *lastTsNs,
|
||||||
Signature: selfSignature,
|
Signature: selfSignature,
|
||||||
ClientId: clientId,
|
ClientId: clientId,
|
||||||
|
|
|
@ -35,7 +35,7 @@ func (s3a *S3ApiServer) subscribeMetaEvents(clientName string, prefix string, la
|
||||||
var clientEpoch int32
|
var clientEpoch int32
|
||||||
util.RetryForever("followIamChanges", func() error {
|
util.RetryForever("followIamChanges", func() error {
|
||||||
clientEpoch++
|
clientEpoch++
|
||||||
return pb.WithFilerClientFollowMetadata(s3a, clientName, s3a.randomClientId, clientEpoch, prefix, &lastTsNs, 0, 0, processEventFn, pb.FatalOnError)
|
return pb.WithFilerClientFollowMetadata(s3a, clientName, s3a.randomClientId, clientEpoch, prefix, nil, &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
|
||||||
|
|
Loading…
Reference in a new issue