lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-10-21 11:07:45 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

add signatures to messages to avoid double processing

Browse Source
This commit is contained in:
Chris Lu
2020-08-28 23:48:48 -07:00
parent 63ad1abcce
commit ca658a97c5
29 changed files with 331 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
weed/server/filer_grpc_server.go
View File

@@ -171,7 +171,7 @@ func (fs *FilerServer) CreateEntry(ctx context.Context, req *filer_pb.CreateEntr
FullPath: util.JoinPath(req.Directory, req.Entry.Name),
Attr: filer2.PbToEntryAttribute(req.Entry.Attributes),
Chunks: chunks,
}, req.OExcl, req.IsFromOtherCluster)
}, req.OExcl, req.IsFromOtherCluster, req.Signatures)
if createErr == nil {
fs.filer.DeleteChunks(garbage)
@@ -235,7 +235,7 @@ func (fs *FilerServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntr
glog.V(3).Infof("UpdateEntry %s: %v", filepath.Join(req.Directory, req.Entry.Name), err)
}
fs.filer.NotifyUpdateEvent(ctx, entry, newEntry, true, req.IsFromOtherCluster)
fs.filer.NotifyUpdateEvent(ctx, entry, newEntry, true, req.IsFromOtherCluster, req.Signatures)
return &filer_pb.UpdateEntryResponse{}, err
}
@@ -312,7 +312,7 @@ func (fs *FilerServer) AppendToEntry(ctx context.Context, req *filer_pb.AppendTo
glog.V(0).Infof("MaybeManifestize: %v", err)
}
err = fs.filer.CreateEntry(context.Background(), entry, false, false)
err = fs.filer.CreateEntry(context.Background(), entry, false, false, nil)
return &filer_pb.AppendToEntryResponse{}, err
}
@@ -321,7 +321,7 @@ func (fs *FilerServer) DeleteEntry(ctx context.Context, req *filer_pb.DeleteEntr
glog.V(4).Infof("DeleteEntry %v", req)
err = fs.filer.DeleteEntryMetaAndData(ctx, util.JoinPath(req.Directory, req.Name), req.IsRecursive, req.IgnoreRecursiveError, req.IsDeleteData, req.IsFromOtherCluster)
err = fs.filer.DeleteEntryMetaAndData(ctx, util.JoinPath(req.Directory, req.Name), req.IsRecursive, req.IgnoreRecursiveError, req.IsDeleteData, req.IsFromOtherCluster, nil)
resp = &filer_pb.DeleteEntryResponse{}
if err != nil {
resp.Error = err.Error()

4
weed/server/filer_grpc_server_rename.go
View File

@@ -110,7 +110,7 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent util.FullPat
Attr: entry.Attr,
Chunks: entry.Chunks,
}
createErr := fs.filer.CreateEntry(ctx, newEntry, false, false)
createErr := fs.filer.CreateEntry(ctx, newEntry, false, false, nil)
if createErr != nil {
return createErr
}
@@ -124,7 +124,7 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent util.FullPat
}
// delete old entry
deleteErr := fs.filer.DeleteEntryMetaAndData(ctx, oldPath, false, false, false, false)
deleteErr := fs.filer.DeleteEntryMetaAndData(ctx, oldPath, false, false, false, false, nil)
if deleteErr != nil {
return deleteErr
}

12
weed/server/filer_grpc_server_sub_meta.go
View File

@@ -24,7 +24,7 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
lastReadTime := time.Unix(0, req.SinceNs)
glog.V(0).Infof(" %v starts to subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
eachEventNotificationFn := eachEventNotificationFn(req, stream, clientName)
eachEventNotificationFn := eachEventNotificationFn(req, stream, clientName, req.Signature)
eachLogEntryFn := eachLogEntryFn(eachEventNotificationFn)
@@ -59,7 +59,7 @@ func (fs *FilerServer) SubscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq
lastReadTime := time.Unix(0, req.SinceNs)
glog.V(0).Infof(" %v local subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
eachEventNotificationFn := eachEventNotificationFn(req, stream, clientName)
eachEventNotificationFn := eachEventNotificationFn(req, stream, clientName, req.Signature)
eachLogEntryFn := eachLogEntryFn(eachEventNotificationFn)
@@ -104,9 +104,15 @@ func eachLogEntryFn(eachEventNotificationFn func(dirPath string, eventNotificati
}
}
func eachEventNotificationFn(req *filer_pb.SubscribeMetadataRequest, stream filer_pb.SeaweedFiler_SubscribeMetadataServer, clientName string) func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error {
func eachEventNotificationFn(req *filer_pb.SubscribeMetadataRequest, stream filer_pb.SeaweedFiler_SubscribeMetadataServer, clientName string, clientSignature int32) func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error {
return func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error {
for _, sig := range eventNotification.Signatures {
if sig == clientSignature && clientSignature != 0 {
return nil
}
}
// get complete path to the file or directory
var entryName string
if eventNotification.OldEntry != nil {

2
weed/server/filer_server_handlers_write.go
View File

@@ -111,7 +111,7 @@ func (fs *FilerServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
objectPath = objectPath[0 : len(objectPath)-1]
}
err := fs.filer.DeleteEntryMetaAndData(context.Background(), util.FullPath(objectPath), isRecursive, ignoreRecursiveError, !skipChunkDeletion, false)
err := fs.filer.DeleteEntryMetaAndData(context.Background(), util.FullPath(objectPath), isRecursive, ignoreRecursiveError, !skipChunkDeletion, false, nil)
if err != nil {
glog.V(1).Infoln("deleting", objectPath, ":", err.Error())
httpStatus := http.StatusInternalServerError

2
weed/server/filer_server_handlers_write_autochunk.go
View File

@@ -172,7 +172,7 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
Size: chunkOffset,
}
if dbErr := fs.filer.CreateEntry(ctx, entry, false, false); dbErr != nil {
if dbErr := fs.filer.CreateEntry(ctx, entry, false, false, nil); dbErr != nil {
fs.filer.DeleteChunks(entry.Chunks)
replyerr = dbErr
filerResult.Error = dbErr.Error()

2
weed/server/filer_server_handlers_write_cipher.go
View File

@@ -80,7 +80,7 @@ func (fs *FilerServer) encrypt(ctx context.Context, w http.ResponseWriter, r *ht
Size: int64(pu.OriginalDataSize),
}
if dbErr := fs.filer.CreateEntry(ctx, entry, false, false); dbErr != nil {
if dbErr := fs.filer.CreateEntry(ctx, entry, false, false, nil); dbErr != nil {
fs.filer.DeleteChunks(entry.Chunks)
err = dbErr
filerResult.Error = dbErr.Error()

11
weed/server/webdav_server.go
View File

@@ -70,6 +70,7 @@ type WebDavFileSystem struct {
filer *filer2.Filer
grpcDialOption grpc.DialOption
chunkCache *chunk_cache.TieredChunkCache
signature int32
}
type FileInfo struct {
@@ -103,6 +104,7 @@ func NewWebDavFileSystem(option *WebDavOption) (webdav.FileSystem, error) {
return &WebDavFileSystem{
option: option,
chunkCache: chunkCache,
signature: util.RandomInt32(),
}, nil
}
@@ -165,6 +167,7 @@ func (fs *WebDavFileSystem) Mkdir(ctx context.Context, fullDirPath string, perm
Gid: fs.option.Gid,
},
},
Signatures: []int32{fs.signature},
}
glog.V(1).Infof("mkdir: %v", request)
@@ -216,6 +219,7 @@ func (fs *WebDavFileSystem) OpenFile(ctx context.Context, fullFilePath string, f
TtlSec: 0,
},
},
Signatures: []int32{fs.signature},
}); err != nil {
return fmt.Errorf("create %s: %v", fullFilePath, err)
}
@@ -255,7 +259,7 @@ func (fs *WebDavFileSystem) removeAll(ctx context.Context, fullFilePath string)
dir, name := util.FullPath(fullFilePath).DirAndName()
return filer_pb.Remove(fs, dir, name, true, false, false, false)
return filer_pb.Remove(fs, dir, name, true, false, false, false, fs.signature)
}
@@ -422,8 +426,9 @@ func (f *WebDavFile) Write(buf []byte) (int, error) {
f.entry.Attributes.Replication = replication
request := &filer_pb.UpdateEntryRequest{
Directory: dir,
Entry: f.entry,
Directory: dir,
Entry: f.entry,
Signatures: []int32{f.fs.signature},
}
if _, err := client.UpdateEntry(ctx, request); err != nil {