mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/chrislusf/seaweedfs synced 2024-07-01 22:56:38 +02:00
Code Issues Projects Releases Wiki Activity

refactoring

Browse source 
some previous chunk etag was using md5, which should be wrong.
This commit is contained in:
Chris Lu 2020-04-30 17:20:44 -07:00
parent 17d5ac4cd3
commit 871efa4fc1
8 changed files with 25 additions and 83 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
weed/command/filer_copy.go
View file

@ -14,9 +14,10 @@ import (
"sync"
"time"
"github.com/chrislusf/seaweedfs/weed/util/grace"
"google.golang.org/grpc"
"github.com/chrislusf/seaweedfs/weed/util/grace"
"github.com/chrislusf/seaweedfs/weed/operation"
"github.com/chrislusf/seaweedfs/weed/pb"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
@ -324,15 +325,7 @@ func (worker *FileCopyWorker) uploadFileAsOne(task FileCopyTask, f *os.File) err
}
fmt.Printf("uploaded %s to %s\n", fileName, targetUrl)
chunks = append(chunks, &filer_pb.FileChunk{
FileId: assignResult.FileId,
Offset: 0,
Size: uint64(uploadResult.Size),
Mtime: time.Now().UnixNano(),
ETag: uploadResult.Md5,
CipherKey: uploadResult.CipherKey,
IsGzipped: uploadResult.Gzip > 0,
})
chunks = append(chunks, uploadResult.ToPbFileChunk(assignResult.FileId, 0))
fmt.Printf("copied %s => http://%s%s%s\n", fileName, worker.filerHost, task.destinationUrlPath, fileName)
}
@ -435,15 +428,8 @@ func (worker *FileCopyWorker) uploadFileInChunks(task FileCopyTask, f *os.File,
uploadError = fmt.Errorf("upload %v to %s result: %v\n", fileName, targetUrl, uploadResult.Error)
return
}
chunksChan <- &filer_pb.FileChunk{
FileId: assignResult.FileId,
Offset: i * chunkSize,
Size: uint64(uploadResult.Size),
Mtime: time.Now().UnixNano(),
ETag: uploadResult.ETag,
CipherKey: uploadResult.CipherKey,
IsGzipped: uploadResult.Gzip > 0,
}
chunksChan <- uploadResult.ToPbFileChunk(assignResult.FileId, i * chunkSize)
fmt.Printf("uploaded %s-%d to %s [%d,%d)\n", fileName, i+1, targetUrl, i*chunkSize, i*chunkSize+int64(uploadResult.Size))
}(i)
}

11
weed/filer2/filer_notify_append.go
View file

@ -38,16 +38,7 @@ func (f *Filer) appendToFile(targetFile string, data []byte) error {
}
// append to existing chunks
chunk := &filer_pb.FileChunk{
FileId: assignResult.Fid,
Offset: offset,
Size: uint64(uploadResult.Size),
Mtime: time.Now().UnixNano(),
ETag: uploadResult.ETag,
CipherKey: uploadResult.CipherKey,
IsGzipped: uploadResult.Gzip > 0,
}
entry.Chunks = append(entry.Chunks, chunk)
entry.Chunks = append(entry.Chunks, uploadResult.ToPbFileChunk(assignResult.Fid, offset))
// update the entry
err = f.CreateEntry(context.Background(), entry, false)

10
weed/filesys/dirty_page.go
View file

@ -187,15 +187,7 @@ func (pages *ContinuousDirtyPages) saveToStorage(reader io.Reader, offset int64,
}
pages.f.wfs.chunkCache.SetChunk(fileId, data)
return &filer_pb.FileChunk{
FileId: fileId,
Offset: offset,
Size: uint64(size),
Mtime: time.Now().UnixNano(),
ETag: uploadResult.ETag,
CipherKey: uploadResult.CipherKey,
IsGzipped: uploadResult.Gzip > 0,
}, nil
return uploadResult.ToPbFileChunk(fileId, offset), nil
}

13
weed/messaging/broker/broker_append.go
View file

@ -3,7 +3,6 @@ package broker
import (
"context"
"fmt"
"time"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/operation"
@ -23,23 +22,13 @@ func (broker *MessageBroker) appendToFile(targetFile string, topicConfig *messag
dir, name := util.FullPath(targetFile).DirAndName()
chunk := &filer_pb.FileChunk{
FileId: assignResult.Fid,
Offset: 0, // needs to be fixed during appending
Size: uint64(uploadResult.Size),
Mtime: time.Now().UnixNano(),
ETag: uploadResult.ETag,
CipherKey: uploadResult.CipherKey,
IsGzipped: uploadResult.Gzip > 0,
}
// append the chunk
if err := broker.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
request := &filer_pb.AppendToEntryRequest{
Directory: dir,
EntryName: name,
Chunks: []*filer_pb.FileChunk{chunk},
Chunks: []*filer_pb.FileChunk{uploadResult.ToPbFileChunk(assignResult.Fid, 0)},
}
_, err := client.AppendToEntry(context.Background(), request)

14
weed/operation/upload_content.go
View file

@ -14,8 +14,10 @@ import (
"net/textproto"
"path/filepath"
"strings"
"time"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/util"
)
@ -31,6 +33,18 @@ type UploadResult struct {
Md5 string `json:"md5,omitempty"`
}
func (uploadResult *UploadResult) ToPbFileChunk(fileId string, offset int64) *filer_pb.FileChunk {
return &filer_pb.FileChunk{
FileId: fileId,
Offset: offset,
Size: uint64(uploadResult.Size),
Mtime: time.Now().UnixNano(),
ETag: uploadResult.ETag,
CipherKey: uploadResult.CipherKey,
IsGzipped: uploadResult.Gzip > 0,
}
}
var (
client *http.Client
)

12
weed/server/filer_server_handlers_write_autochunk.go
View file

@ -119,17 +119,7 @@ func (fs *FilerServer) doAutoChunk(ctx context.Context, w http.ResponseWriter, r
}
// Save to chunk manifest structure
fileChunks = append(fileChunks,
&filer_pb.FileChunk{
FileId: fileId,
Offset: chunkOffset,
Size: uint64(uploadResult.Size),
Mtime: time.Now().UnixNano(),
ETag: uploadResult.ETag,
CipherKey: uploadResult.CipherKey,
IsGzipped: uploadResult.Gzip > 0,
},
)
fileChunks = append(fileChunks, uploadResult.ToPbFileChunk(fileId, chunkOffset))
glog.V(4).Infof("uploaded %s chunk %d to %s [%d,%d) of %d", fileName, len(fileChunks), fileId, chunkOffset, chunkOffset+int64(uploadResult.Size), contentLength)

12
weed/server/filer_server_handlers_write_cipher.go
View file

@ -46,17 +46,7 @@ func (fs *FilerServer) encrypt(ctx context.Context, w http.ResponseWriter, r *ht
}
// Save to chunk manifest structure
fileChunks := []*filer_pb.FileChunk{
{
FileId: fileId,
Offset: 0,
Size: uint64(uploadResult.Size),
Mtime: time.Now().UnixNano(),
ETag: uploadResult.Md5,
CipherKey: uploadResult.CipherKey,
IsGzipped: uploadResult.Gzip > 0,
},
}
fileChunks := []*filer_pb.FileChunk{uploadResult.ToPbFileChunk(fileId, 0)}
// fmt.Printf("uploaded: %+v\n", uploadResult)

12
weed/server/webdav_server.go
View file

@ -418,17 +418,7 @@ func (f *WebDavFile) Write(buf []byte) (int, error) {
return 0, fmt.Errorf("upload result: %v", uploadResult.Error)
}
chunk := &filer_pb.FileChunk{
FileId: fileId,
Offset: f.off,
Size: uint64(len(buf)),
Mtime: time.Now().UnixNano(),
ETag: uploadResult.ETag,
CipherKey: uploadResult.CipherKey,
IsGzipped: uploadResult.Gzip > 0,
}
f.entry.Chunks = append(f.entry.Chunks, chunk)
f.entry.Chunks = append(f.entry.Chunks, uploadResult.ToPbFileChunk(fileId, f.off))
err = f.fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
f.entry.Attributes.Mtime = time.Now().Unix()