From 31922b2bf232af8de55adde44138437aeefea1c8 Mon Sep 17 00:00:00 2001 From: chrislu Date: Tue, 20 Sep 2022 09:30:05 -0700 Subject: [PATCH] s3 to watch specific directories --- weed/mount/meta_cache/meta_cache_subscribe.go | 2 +- weed/pb/filer_pb_tail.go | 13 +++++-------- weed/s3api/auth_credentials_subscribe.go | 2 +- 3 files changed, 7 insertions(+), 10 deletions(-) diff --git a/weed/mount/meta_cache/meta_cache_subscribe.go b/weed/mount/meta_cache/meta_cache_subscribe.go index 241777371..f52660a42 100644 --- a/weed/mount/meta_cache/meta_cache_subscribe.go +++ b/weed/mount/meta_cache/meta_cache_subscribe.go @@ -60,7 +60,7 @@ func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.Fil var clientEpoch int32 util.RetryForever("followMetaUpdates", func() error { 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 { glog.Errorf("follow metadata updates: %v", err) return true diff --git a/weed/pb/filer_pb_tail.go b/weed/pb/filer_pb_tail.go index f7e09b386..b8a74f1d5 100644 --- a/weed/pb/filer_pb_tail.go +++ b/weed/pb/filer_pb_tail.go @@ -25,20 +25,16 @@ func FollowMetadata(filerAddress ServerAddress, grpcDialOption grpc.DialOption, pathPrefix string, additionalPathPrefixes []string, lastTsNs int64, untilTsNs int64, selfSignature int32, processEventFn ProcessMetadataFunc, eventErrorType EventErrorType) error { - err := WithFilerClient(true, filerAddress, grpcDialOption, makeSubscribeMetadataFunc( - clientName, clientId, clientEpoch, - pathPrefix, additionalPathPrefixes, &lastTsNs, untilTsNs, selfSignature, processEventFn, eventErrorType)) + err := WithFilerClient(true, filerAddress, grpcDialOption, makeSubscribeMetadataFunc(clientName, clientId, clientEpoch, pathPrefix, additionalPathPrefixes, nil, &lastTsNs, untilTsNs, selfSignature, processEventFn, eventErrorType)) if err != nil { return fmt.Errorf("subscribing filer meta change: %v", err) } return err } -func WithFilerClientFollowMetadata(filerClient filer_pb.FilerClient, - clientName string, clientId int32, clientEpoch int32, pathPrefix string, lastTsNs *int64, untilTsNs int64, selfSignature int32, - processEventFn ProcessMetadataFunc, eventErrorType EventErrorType) error { +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 { - 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 { return fmt.Errorf("subscribing filer meta change: %v", err) } @@ -46,7 +42,7 @@ func WithFilerClientFollowMetadata(filerClient filer_pb.FilerClient, 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 { ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -54,6 +50,7 @@ func makeSubscribeMetadataFunc(clientName string, clientId int32, clientEpoch in ClientName: clientName, PathPrefix: pathPrefix, PathPrefixes: additionalPathPrefixes, + Directories: directoriesToWatch, SinceNs: *lastTsNs, Signature: selfSignature, ClientId: clientId, diff --git a/weed/s3api/auth_credentials_subscribe.go b/weed/s3api/auth_credentials_subscribe.go index f6b5b3d80..d567bb5d7 100644 --- a/weed/s3api/auth_credentials_subscribe.go +++ b/weed/s3api/auth_credentials_subscribe.go @@ -35,7 +35,7 @@ func (s3a *S3ApiServer) subscribeMetaEvents(clientName string, prefix string, la var clientEpoch int32 util.RetryForever("followIamChanges", func() error { 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 { glog.V(0).Infof("iam follow metadata changes: %v", err) return true