lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-09-24 02:13:35 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

filer sink retries reading file chunks, skipping missing chunks

Browse Source 
if the file chunk is not available during replication time, the file is skipped
This commit is contained in:
chrislu
2022-12-19 11:31:58 -08:00
parent 3fc3e7083c
commit 6c7fe40305
2 changed files with 15 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
weed/replication/sink/filersink/fetch_write.go
View File

@@ -27,12 +27,16 @@ func (fs *FilerSink) replicateChunks(sourceChunks []*filer_pb.FileChunk, path st
index, source := chunkIndex, sourceChunk index, source := chunkIndex, sourceChunk
fs.executor.Execute(func() { fs.executor.Execute(func() {
defer wg.Done() defer wg.Done()
util.Retry("replicate chunks", func() error {
replicatedChunk, e := fs.replicateOneChunk(source, path) replicatedChunk, e := fs.replicateOneChunk(source, path)
if e != nil { if e != nil {
err = e err = e
return return e
} }
replicatedChunks[index] = replicatedChunk replicatedChunks[index] = replicatedChunk
err = nil
return nil
})
}) })
} }
wg.Wait() wg.Wait()

8
weed/replication/sink/filersink/filer_sink.go
View File

@@ -112,7 +112,7 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [
Directory: dir, Directory: dir,
Name: name, Name: name,
} }
glog.V(1).Infof("lookup: %v", lookupRequest) // glog.V(1).Infof("lookup: %v", lookupRequest)
if resp, err := filer_pb.LookupEntry(client, lookupRequest); err == nil { if resp, err := filer_pb.LookupEntry(client, lookupRequest); err == nil {
if filer.ETag(resp.Entry) == filer.ETag(entry) { if filer.ETag(resp.Entry) == filer.ETag(entry) {
glog.V(3).Infof("already replicated %s", key) glog.V(3).Infof("already replicated %s", key)
@@ -125,9 +125,10 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [
if err != nil { if err != nil {
// only warning here since the source chunk may have been deleted already // only warning here since the source chunk may have been deleted already
glog.Warningf("replicate entry chunks %s: %v", key, err) glog.Warningf("replicate entry chunks %s: %v", key, err)
return nil
} }
glog.V(4).Infof("replicated %s %+v ===> %+v", key, entry.GetChunks(), replicatedChunks) // glog.V(4).Infof("replicated %s %+v ===> %+v", key, entry.GetChunks(), replicatedChunks)
request := &filer_pb.CreateEntryRequest{ request := &filer_pb.CreateEntryRequest{
Directory: dir, Directory: dir,
@@ -205,7 +206,8 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent
// replicate the chunks that are new in the source // replicate the chunks that are new in the source
replicatedChunks, err := fs.replicateChunks(newChunks, key) replicatedChunks, err := fs.replicateChunks(newChunks, key)
if err != nil { if err != nil {
return true, fmt.Errorf("replicate %s chunks error: %v", key, err) glog.Warningf("replicate entry chunks %s: %v", key, err)
return true, nil
} }
existingEntry.Chunks = append(existingEntry.GetChunks(), replicatedChunks...) existingEntry.Chunks = append(existingEntry.GetChunks(), replicatedChunks...)
existingEntry.Attributes = newEntry.Attributes existingEntry.Attributes = newEntry.Attributes