From 31ed352ab638027e940ff92af55c0c6db0dc3309 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 25 Sep 2018 09:27:03 -0700 Subject: [PATCH] replication handle cases when entry already exists --- weed/replication/replicator.go | 16 ++++++++++- weed/replication/sink/filersink/filer_sink.go | 28 +++++++++++++------ weed/replication/sink/replication_sink.go | 3 +- 3 files changed, 36 insertions(+), 11 deletions(-) diff --git a/weed/replication/replicator.go b/weed/replication/replicator.go index 215be5992..834da6217 100644 --- a/weed/replication/replicator.go +++ b/weed/replication/replicator.go @@ -9,6 +9,7 @@ import ( "github.com/chrislusf/seaweedfs/weed/replication/sink/filersink" "github.com/chrislusf/seaweedfs/weed/replication/source" "github.com/chrislusf/seaweedfs/weed/util" + "github.com/chrislusf/seaweedfs/weed/glog" ) type Replicator struct { @@ -43,5 +44,18 @@ func (r *Replicator) Replicate(key string, message *filer_pb.EventNotification) if message.OldEntry == nil && message.NewEntry != nil { return r.sink.CreateEntry(key, message.NewEntry) } - return r.sink.UpdateEntry(key, message.OldEntry, message.NewEntry, message.DeleteChunks) + if existingEntry, err := r.sink.LookupEntry(key); err == nil { + if message.OldEntry == nil && message.NewEntry == nil { + glog.V(0).Infof("message %+v existingEntry: %+v", message, existingEntry) + return r.sink.DeleteEntry(key, existingEntry, true) + } + return r.sink.UpdateEntry(key, message.OldEntry, message.NewEntry, existingEntry, message.DeleteChunks) + } + + glog.V(0).Infof("key:%s, message %+v", key, message) + if message.OldEntry == nil && message.NewEntry == nil { + return nil + } + + return r.sink.CreateEntry(key, message.NewEntry) } diff --git a/weed/replication/sink/filersink/filer_sink.go b/weed/replication/sink/filersink/filer_sink.go index 1cbf52864..c98c99f34 100644 --- a/weed/replication/sink/filersink/filer_sink.go +++ b/weed/replication/sink/filersink/filer_sink.go @@ -121,14 +121,13 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry) error { }) } -func (fs *FilerSink) UpdateEntry(key string, oldEntry, newEntry *filer_pb.Entry, deleteIncludeChunks bool) (err error) { +func (fs *FilerSink) LookupEntry(key string) (entry *filer_pb.Entry, err error) { ctx := context.Background() dir, name := filer2.FullPath(key).DirAndName() // read existing entry - var entry *filer_pb.Entry err = fs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error { request := &filer_pb.LookupDirectoryEntryRequest{ @@ -136,7 +135,7 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry, newEntry *filer_pb.Entry, Name: name, } - glog.V(4).Infof("lookup directory entry: %v", request) + glog.V(4).Infof("lookup entry: %v", request) resp, err := client.LookupDirectoryEntry(ctx, request) if err != nil { glog.V(0).Infof("lookup %s: %v", key, err) @@ -149,10 +148,21 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry, newEntry *filer_pb.Entry, }) if err != nil { - return fmt.Errorf("lookup when updating %s: %v", key, err) + return nil, fmt.Errorf("lookup %s: %v", key, err) } - if filer2.ETag(newEntry.Chunks) == filer2.ETag(entry.Chunks) { + return entry, nil +} + +func (fs *FilerSink) UpdateEntry(key string, oldEntry, newEntry, existingEntry *filer_pb.Entry, deleteIncludeChunks bool) (err error) { + + ctx := context.Background() + + dir, _ := filer2.FullPath(key).DirAndName() + + glog.V(0).Infof("oldEntry %+v, newEntry %+v, existingEntry: %+v", oldEntry, newEntry, existingEntry) + + if filer2.ETag(newEntry.Chunks) == filer2.ETag(existingEntry.Chunks) { // skip if no change // this usually happens when retrying the replication glog.V(0).Infof("already replicated %s", key) @@ -163,7 +173,7 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry, newEntry *filer_pb.Entry, // delete the chunks that are deleted from the source if deleteIncludeChunks { // remove the deleted chunks. Actual data deletion happens in filer UpdateEntry FindUnusedFileChunks - entry.Chunks = minusChunks(entry.Chunks, deletedChunks) + existingEntry.Chunks = minusChunks(existingEntry.Chunks, deletedChunks) } // replicate the chunks that are new in the source @@ -171,7 +181,7 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry, newEntry *filer_pb.Entry, if err != nil { return fmt.Errorf("replicte %s chunks error: %v", key, err) } - entry.Chunks = append(entry.Chunks, replicatedChunks...) + existingEntry.Chunks = append(existingEntry.Chunks, replicatedChunks...) } // save updated meta data @@ -179,11 +189,11 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry, newEntry *filer_pb.Entry, request := &filer_pb.UpdateEntryRequest{ Directory: dir, - Entry: entry, + Entry: existingEntry, } if _, err := client.UpdateEntry(ctx, request); err != nil { - return fmt.Errorf("update entry %s: %v", key, err) + return fmt.Errorf("update existingEntry %s: %v", key, err) } return nil diff --git a/weed/replication/sink/replication_sink.go b/weed/replication/sink/replication_sink.go index bb4a8aa83..c33f3251b 100644 --- a/weed/replication/sink/replication_sink.go +++ b/weed/replication/sink/replication_sink.go @@ -8,7 +8,8 @@ import ( type ReplicationSink interface { DeleteEntry(key string, entry *filer_pb.Entry, deleteIncludeChunks bool) error CreateEntry(key string, entry *filer_pb.Entry) error - UpdateEntry(key string, oldEntry, newEntry *filer_pb.Entry, deleteIncludeChunks bool) error + UpdateEntry(key string, oldEntry, newEntry, existingEntry *filer_pb.Entry, deleteIncludeChunks bool) error + LookupEntry(key string) (entry *filer_pb.Entry, err error) GetSinkToDirectory() string SetSourceFiler(s *source.FilerSource) }