mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-11-24 19:19:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactoring

Browse Source
This commit is contained in:
Chris Lu 2020-11-15 14:41:56 -08:00
parent 590f02179d
commit 500bcab953
6 changed files with 82 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
weed/filer/filer_conf.go
View File

@ -51,7 +51,6 @@ func (fc *FilerConf) loadFromChunks(filer *Filer, chunks []*filer_pb.FileChunk)
return fc.loadFromBytes(data) return fc.loadFromBytes(data)
} }
func (fc *FilerConf) loadFromBytes(data []byte) (err error) { func (fc *FilerConf) loadFromBytes(data []byte) (err error) {
conf := &filer_pb.FilerConf{} conf := &filer_pb.FilerConf{}
err = proto.UnmarshalText(string(data), conf) err = proto.UnmarshalText(string(data), conf)

16
weed/filer/storage_option.go Normal file
View File

@ -0,0 +1,16 @@
package filer
import "github.com/chrislusf/seaweedfs/weed/storage/needle"
type StorageOption struct {
Replication string
Collection string
DataCenter string
Rack string
TtlSeconds int32
Fsync bool
}
func (so *StorageOption) TtlString() string {
return needle.SecondsToTTL(so.TtlSeconds)
}

21
weed/server/filer_grpc_server.go
View File

@ -257,7 +257,15 @@ func (fs *FilerServer) cleanupChunks(existingEntry *filer.Entry, newEntry *filer
garbage = append(garbage, coveredChunks...) garbage = append(garbage, coveredChunks...)
if newEntry.Attributes != nil { if newEntry.Attributes != nil {
chunks, err = filer.MaybeManifestize(fs.saveAsChunk(newEntry.Attributes.Replication, newEntry.Attributes.Collection, "", "", needle.SecondsToTTL(newEntry.Attributes.TtlSec), false), chunks) so := &filer.StorageOption{
Replication: newEntry.Attributes.Replication,
Collection: newEntry.Attributes.Collection,
DataCenter: "",
Rack: "",
TtlSeconds: newEntry.Attributes.TtlSec,
Fsync: false,
}
chunks, err = filer.MaybeManifestize(fs.saveAsChunk(so), chunks)
if err != nil { if err != nil {
// not good, but should be ok // not good, but should be ok
glog.V(0).Infof("MaybeManifestize: %v", err) glog.V(0).Infof("MaybeManifestize: %v", err)
@ -297,8 +305,15 @@ func (fs *FilerServer) AppendToEntry(ctx context.Context, req *filer_pb.AppendTo
} }
entry.Chunks = append(entry.Chunks, req.Chunks...) entry.Chunks = append(entry.Chunks, req.Chunks...)
so := &filer.StorageOption{
entry.Chunks, err = filer.MaybeManifestize(fs.saveAsChunk(entry.Replication, entry.Collection, "", "", needle.SecondsToTTL(entry.TtlSec), false), entry.Chunks) Replication: entry.Replication,
Collection: entry.Collection,
DataCenter: "",
Rack: "",
TtlSeconds: entry.TtlSec,
Fsync: false,
}
entry.Chunks, err = filer.MaybeManifestize(fs.saveAsChunk(so), entry.Chunks)
if err != nil { if err != nil {
// not good, but should be ok // not good, but should be ok
glog.V(0).Infof("MaybeManifestize: %v", err) glog.V(0).Infof("MaybeManifestize: %v", err)

33
weed/server/filer_server_handlers_write.go
View File

@ -7,6 +7,7 @@ import (
"strings" "strings"
"time" "time"
"github.com/chrislusf/seaweedfs/weed/filer"
"github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/operation" "github.com/chrislusf/seaweedfs/weed/operation"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb" "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
@ -29,7 +30,7 @@ type FilerPostResult struct {
Url string `json:"url,omitempty"` Url string `json:"url,omitempty"`
} }
func (fs *FilerServer) assignNewFileInfo(replication, collection, dataCenter, rack, ttlString string, fsync bool) (fileId, urlLocation string, auth security.EncodedJwt, err error) { func (fs *FilerServer) assignNewFileInfo(so *filer.StorageOption) (fileId, urlLocation string, auth security.EncodedJwt, err error) {
stats.FilerRequestCounter.WithLabelValues("assign").Inc() stats.FilerRequestCounter.WithLabelValues("assign").Inc()
start := time.Now() start := time.Now()
@ -37,18 +38,19 @@ func (fs *FilerServer) assignNewFileInfo(replication, collection, dataCenter, ra
ar := &operation.VolumeAssignRequest{ ar := &operation.VolumeAssignRequest{
Count: 1, Count: 1,
Replication: replication, Replication: so.Replication,
Collection: collection, Collection: so.Collection,
Ttl: ttlString, Ttl: so.TtlString(),
DataCenter: dataCenter, DataCenter: so.DataCenter,
Rack: so.Rack,
} }
var altRequest *operation.VolumeAssignRequest var altRequest *operation.VolumeAssignRequest
if dataCenter != "" || rack != "" { if so.DataCenter != "" || so.Rack != "" {
altRequest = &operation.VolumeAssignRequest{ altRequest = &operation.VolumeAssignRequest{
Count: 1, Count: 1,
Replication: replication, Replication: so.Replication,
Collection: collection, Collection: so.Collection,
Ttl: ttlString, Ttl: so.TtlString(),
DataCenter: "", DataCenter: "",
Rack: "", Rack: "",
} }
@ -62,7 +64,7 @@ func (fs *FilerServer) assignNewFileInfo(replication, collection, dataCenter, ra
} }
fileId = assignResult.Fid fileId = assignResult.Fid
urlLocation = "http://" + assignResult.Url + "/" + assignResult.Fid urlLocation = "http://" + assignResult.Url + "/" + assignResult.Fid
if fsync { if so.Fsync {
urlLocation += "?fsync=true" urlLocation += "?fsync=true"
} }
auth = assignResult.Auth auth = assignResult.Auth
@ -92,7 +94,16 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
ttlSeconds = int32(ttl.Minutes()) * 60 ttlSeconds = int32(ttl.Minutes()) * 60
} }
fs.autoChunk(ctx, w, r, replication, collection, dataCenter, rack, ttlSeconds, ttlString, fsync) so := &filer.StorageOption{
Replication: replication,
Collection: collection,
DataCenter: dataCenter,
Rack: rack,
TtlSeconds: ttlSeconds,
Fsync: fsync,
}
fs.autoChunk(ctx, w, r, so)
} }

40
weed/server/filer_server_handlers_write_autochunk.go
View File

@ -25,7 +25,7 @@ import (
"github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/util"
) )
func (fs *FilerServer) autoChunk(ctx context.Context, w http.ResponseWriter, r *http.Request, replication string, collection string, dataCenter string, rack string, ttlSec int32, ttlString string, fsync bool) { func (fs *FilerServer) autoChunk(ctx context.Context, w http.ResponseWriter, r *http.Request, so *filer.StorageOption) {
// autoChunking can be set at the command-line level or as a query param. Query param overrides command-line // autoChunking can be set at the command-line level or as a query param. Query param overrides command-line
query := r.URL.Query() query := r.URL.Query()
@ -51,10 +51,10 @@ func (fs *FilerServer) autoChunk(ctx context.Context, w http.ResponseWriter, r *
if r.Header.Get("Content-Type") == "" && strings.HasSuffix(r.URL.Path, "/") { if r.Header.Get("Content-Type") == "" && strings.HasSuffix(r.URL.Path, "/") {
reply, err = fs.mkdir(ctx, w, r) reply, err = fs.mkdir(ctx, w, r)
} else { } else {
reply, md5bytes, err = fs.doPostAutoChunk(ctx, w, r, chunkSize, replication, collection, dataCenter, rack, ttlSec, ttlString, fsync) reply, md5bytes, err = fs.doPostAutoChunk(ctx, w, r, chunkSize, so)
} }
} else { } else {
reply, md5bytes, err = fs.doPutAutoChunk(ctx, w, r, chunkSize, replication, collection, dataCenter, rack, ttlSec, ttlString, fsync) reply, md5bytes, err = fs.doPutAutoChunk(ctx, w, r, chunkSize, so)
} }
if err != nil { if err != nil {
writeJsonError(w, r, http.StatusInternalServerError, err) writeJsonError(w, r, http.StatusInternalServerError, err)
@ -66,7 +66,7 @@ func (fs *FilerServer) autoChunk(ctx context.Context, w http.ResponseWriter, r *
} }
} }
func (fs *FilerServer) doPostAutoChunk(ctx context.Context, w http.ResponseWriter, r *http.Request, chunkSize int32, replication string, collection string, dataCenter string, rack string, ttlSec int32, ttlString string, fsync bool) (filerResult *FilerPostResult, md5bytes []byte, replyerr error) { func (fs *FilerServer) doPostAutoChunk(ctx context.Context, w http.ResponseWriter, r *http.Request, chunkSize int32, so *filer.StorageOption) (filerResult *FilerPostResult, md5bytes []byte, replyerr error) {
multipartReader, multipartReaderErr := r.MultipartReader() multipartReader, multipartReaderErr := r.MultipartReader()
if multipartReaderErr != nil { if multipartReaderErr != nil {
@ -87,46 +87,46 @@ func (fs *FilerServer) doPostAutoChunk(ctx context.Context, w http.ResponseWrite
contentType = "" contentType = ""
} }
fileChunks, md5Hash, chunkOffset, err := fs.uploadReaderToChunks(w, r, part1, chunkSize, replication, collection, dataCenter, rack, ttlString, fileName, contentType, fsync) fileChunks, md5Hash, chunkOffset, err := fs.uploadReaderToChunks(w, r, part1, chunkSize, fileName, contentType, so)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
fileChunks, replyerr = filer.MaybeManifestize(fs.saveAsChunk(replication, collection, dataCenter, rack, ttlString, fsync), fileChunks) fileChunks, replyerr = filer.MaybeManifestize(fs.saveAsChunk(so), fileChunks)
if replyerr != nil { if replyerr != nil {
glog.V(0).Infof("manifestize %s: %v", r.RequestURI, replyerr) glog.V(0).Infof("manifestize %s: %v", r.RequestURI, replyerr)
return return
} }
md5bytes = md5Hash.Sum(nil) md5bytes = md5Hash.Sum(nil)
filerResult, replyerr = fs.saveMetaData(ctx, r, fileName, replication, collection, ttlSec, contentType, md5bytes, fileChunks, chunkOffset) filerResult, replyerr = fs.saveMetaData(ctx, r, fileName, contentType, so, md5bytes, fileChunks, chunkOffset)
return return
} }
func (fs *FilerServer) doPutAutoChunk(ctx context.Context, w http.ResponseWriter, r *http.Request, chunkSize int32, replication string, collection string, dataCenter string, rack string, ttlSec int32, ttlString string, fsync bool) (filerResult *FilerPostResult, md5bytes []byte, replyerr error) { func (fs *FilerServer) doPutAutoChunk(ctx context.Context, w http.ResponseWriter, r *http.Request, chunkSize int32, so *filer.StorageOption) (filerResult *FilerPostResult, md5bytes []byte, replyerr error) {
fileName := "" fileName := ""
contentType := "" contentType := ""
fileChunks, md5Hash, chunkOffset, err := fs.uploadReaderToChunks(w, r, r.Body, chunkSize, replication, collection, dataCenter, rack, ttlString, fileName, contentType, fsync) fileChunks, md5Hash, chunkOffset, err := fs.uploadReaderToChunks(w, r, r.Body, chunkSize, fileName, contentType, so)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
fileChunks, replyerr = filer.MaybeManifestize(fs.saveAsChunk(replication, collection, dataCenter, rack, ttlString, fsync), fileChunks) fileChunks, replyerr = filer.MaybeManifestize(fs.saveAsChunk(so), fileChunks)
if replyerr != nil { if replyerr != nil {
glog.V(0).Infof("manifestize %s: %v", r.RequestURI, replyerr) glog.V(0).Infof("manifestize %s: %v", r.RequestURI, replyerr)
return return
} }
md5bytes = md5Hash.Sum(nil) md5bytes = md5Hash.Sum(nil)
filerResult, replyerr = fs.saveMetaData(ctx, r, fileName, replication, collection, ttlSec, contentType, md5bytes, fileChunks, chunkOffset) filerResult, replyerr = fs.saveMetaData(ctx, r, fileName, contentType, so, md5bytes, fileChunks, chunkOffset)
return return
} }
func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileName string, replication string, collection string, ttlSec int32, contentType string, md5bytes []byte, fileChunks []*filer_pb.FileChunk, chunkOffset int64) (filerResult *FilerPostResult, replyerr error) { func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileName string, contentType string, so *filer.StorageOption, md5bytes []byte, fileChunks []*filer_pb.FileChunk, chunkOffset int64) (filerResult *FilerPostResult, replyerr error) {
// detect file mode // detect file mode
modeStr := r.URL.Query().Get("mode") modeStr := r.URL.Query().Get("mode")
@ -163,9 +163,9 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
Mode: os.FileMode(mode), Mode: os.FileMode(mode),
Uid: OS_UID, Uid: OS_UID,
Gid: OS_GID, Gid: OS_GID,
Replication: replication, Replication: so.Replication,
Collection: collection, Collection: so.Collection,
TtlSec: ttlSec, TtlSec: so.TtlSeconds,
Mime: contentType, Mime: contentType,
Md5: md5bytes, Md5: md5bytes,
FileSize: uint64(chunkOffset), FileSize: uint64(chunkOffset),
@ -199,7 +199,7 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
return filerResult, replyerr return filerResult, replyerr
} }
func (fs *FilerServer) uploadReaderToChunks(w http.ResponseWriter, r *http.Request, reader io.Reader, chunkSize int32, replication string, collection string, dataCenter string, rack string, ttlString string, fileName string, contentType string, fsync bool) ([]*filer_pb.FileChunk, hash.Hash, int64, error) { func (fs *FilerServer) uploadReaderToChunks(w http.ResponseWriter, r *http.Request, reader io.Reader, chunkSize int32, fileName, contentType string, so *filer.StorageOption) ([]*filer_pb.FileChunk, hash.Hash, int64, error) {
var fileChunks []*filer_pb.FileChunk var fileChunks []*filer_pb.FileChunk
md5Hash := md5.New() md5Hash := md5.New()
@ -211,7 +211,7 @@ func (fs *FilerServer) uploadReaderToChunks(w http.ResponseWriter, r *http.Reque
limitedReader := io.LimitReader(partReader, int64(chunkSize)) limitedReader := io.LimitReader(partReader, int64(chunkSize))
// assign one file id for one chunk // assign one file id for one chunk
fileId, urlLocation, auth, assignErr := fs.assignNewFileInfo(replication, collection, dataCenter, rack, ttlString, fsync) fileId, urlLocation, auth, assignErr := fs.assignNewFileInfo(so)
if assignErr != nil { if assignErr != nil {
return nil, nil, 0, assignErr return nil, nil, 0, assignErr
} }
@ -255,11 +255,11 @@ func (fs *FilerServer) doUpload(urlLocation string, w http.ResponseWriter, r *ht
return uploadResult, err return uploadResult, err
} }
func (fs *FilerServer) saveAsChunk(replication string, collection string, dataCenter string, rack string, ttlString string, fsync bool) filer.SaveDataAsChunkFunctionType { func (fs *FilerServer) saveAsChunk(so *filer.StorageOption) filer.SaveDataAsChunkFunctionType {
return func(reader io.Reader, name string, offset int64) (*filer_pb.FileChunk, string, string, error) { return func(reader io.Reader, name string, offset int64) (*filer_pb.FileChunk, string, string, error) {
// assign one file id for one chunk // assign one file id for one chunk
fileId, urlLocation, auth, assignErr := fs.assignNewFileInfo(replication, collection, dataCenter, rack, ttlString, fsync) fileId, urlLocation, auth, assignErr := fs.assignNewFileInfo(so)
if assignErr != nil { if assignErr != nil {
return nil, "", "", assignErr return nil, "", "", assignErr
} }
@ -270,7 +270,7 @@ func (fs *FilerServer) saveAsChunk(replication string, collection string, dataCe
return nil, "", "", uploadErr return nil, "", "", uploadErr
} }
return uploadResult.ToPbFileChunk(fileId, offset), collection, replication, nil return uploadResult.ToPbFileChunk(fileId, offset), so.Collection, so.Replication, nil
} }
} }

12
weed/server/filer_server_handlers_write_cipher.go
View File

@ -16,12 +16,12 @@ import (
) )
// handling single chunk POST or PUT upload // handling single chunk POST or PUT upload
func (fs *FilerServer) encrypt(ctx context.Context, w http.ResponseWriter, r *http.Request, replication string, collection string, dataCenter string, rack string, ttlSeconds int32, ttlString string, fsync bool) (filerResult *FilerPostResult, err error) { func (fs *FilerServer) encrypt(ctx context.Context, w http.ResponseWriter, r *http.Request, so *filer.StorageOption) (filerResult *FilerPostResult, err error) {
fileId, urlLocation, auth, err := fs.assignNewFileInfo(replication, collection, dataCenter, rack, ttlString, fsync) fileId, urlLocation, auth, err := fs.assignNewFileInfo(so)
if err != nil || fileId == "" || urlLocation == "" { if err != nil || fileId == "" || urlLocation == "" {
return nil, fmt.Errorf("fail to allocate volume for %s, collection:%s, datacenter:%s", r.URL.Path, collection, dataCenter) return nil, fmt.Errorf("fail to allocate volume for %s, collection:%s, datacenter:%s", r.URL.Path, so.Collection, so.DataCenter)
} }
glog.V(4).Infof("write %s to %v", r.URL.Path, urlLocation) glog.V(4).Infof("write %s to %v", r.URL.Path, urlLocation)
@ -65,9 +65,9 @@ func (fs *FilerServer) encrypt(ctx context.Context, w http.ResponseWriter, r *ht
Mode: 0660, Mode: 0660,
Uid: OS_UID, Uid: OS_UID,
Gid: OS_GID, Gid: OS_GID,
Replication: replication, Replication: so.Replication,
Collection: collection, Collection: so.Collection,
TtlSec: ttlSeconds, TtlSec: so.TtlSeconds,
Mime: pu.MimeType, Mime: pu.MimeType,
Md5: util.Base64Md5ToBytes(pu.ContentMd5), Md5: util.Base64Md5ToBytes(pu.ContentMd5),
}, },