1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
|
package meta_cache
import (
"context"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/util"
"strings"
)
type MetadataFollower struct {
PathPrefixToWatch string
ProcessEventFn func(resp *filer_pb.SubscribeMetadataResponse) error
}
func mergeProcessors(mainProcessor func(resp *filer_pb.SubscribeMetadataResponse) error, followers ...*MetadataFollower) func(resp *filer_pb.SubscribeMetadataResponse) error {
return func(resp *filer_pb.SubscribeMetadataResponse) error {
// build the full path
entry := resp.EventNotification.NewEntry
if entry == nil {
entry = resp.EventNotification.OldEntry
}
if entry != nil {
dir := resp.Directory
if resp.EventNotification.NewParentPath != "" {
dir = resp.EventNotification.NewParentPath
}
fp := util.NewFullPath(dir, entry.Name)
for _, follower := range followers {
if strings.HasPrefix(string(fp), follower.PathPrefixToWatch) {
if err := follower.ProcessEventFn(resp); err != nil {
return err
}
}
}
}
return mainProcessor(resp)
}
}
func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.FilerClient, dir string, lastTsNs int64, followers ...*MetadataFollower) error {
var prefixes []string
for _, follower := range followers {
prefixes = append(prefixes, follower.PathPrefixToWatch)
}
processEventFn := func(resp *filer_pb.SubscribeMetadataResponse) error {
message := resp.EventNotification
for _, sig := range message.Signatures {
if sig == selfSignature && selfSignature != 0 {
return nil
}
}
dir := resp.Directory
var oldPath util.FullPath
var newEntry *filer.Entry
if message.OldEntry != nil {
oldPath = util.NewFullPath(dir, message.OldEntry.Name)
glog.V(4).Infof("deleting %v", oldPath)
}
if message.NewEntry != nil {
if message.NewParentPath != "" {
dir = message.NewParentPath
}
key := util.NewFullPath(dir, message.NewEntry.Name)
glog.V(4).Infof("creating %v", key)
newEntry = filer.FromPbEntry(dir, message.NewEntry)
}
err := mc.AtomicUpdateEntryFromFiler(context.Background(), oldPath, newEntry)
if err == nil {
if message.OldEntry != nil && message.NewEntry != nil {
oldKey := util.NewFullPath(resp.Directory, message.OldEntry.Name)
mc.invalidateFunc(oldKey, message.OldEntry)
if message.OldEntry.Name != message.NewEntry.Name {
newKey := util.NewFullPath(dir, message.NewEntry.Name)
mc.invalidateFunc(newKey, message.NewEntry)
}
} else if filer_pb.IsCreate(resp) {
// no need to invalidate
} else if filer_pb.IsDelete(resp) {
oldKey := util.NewFullPath(resp.Directory, message.OldEntry.Name)
mc.invalidateFunc(oldKey, message.OldEntry)
}
}
return err
}
prefix := dir
if !strings.HasSuffix(prefix, "/") {
prefix = prefix + "/"
}
metadataFollowOption := &pb.MetadataFollowOption{
ClientName: "mount",
ClientId: selfSignature,
ClientEpoch: 1,
SelfSignature: selfSignature,
PathPrefix: prefix,
AdditionalPathPrefixes: prefixes,
DirectoriesToWatch: nil,
StartTsNs: lastTsNs,
StopTsNs: 0,
EventErrorType: pb.FatalOnError,
}
util.RetryUntil("followMetaUpdates", func() error {
metadataFollowOption.ClientEpoch++
return pb.WithFilerClientFollowMetadata(client, metadataFollowOption, mergeProcessors(processEventFn, followers...))
}, func(err error) bool {
glog.Errorf("follow metadata updates: %v", err)
return true
})
return nil
}
|
