diff --git a/weed/filer/filer_conf.go b/weed/filer/filer_conf.go index 737b1aeea..0a86192ec 100644 --- a/weed/filer/filer_conf.go +++ b/weed/filer/filer_conf.go @@ -51,7 +51,6 @@ func (fc *FilerConf) loadFromChunks(filer *Filer, chunks []*filer_pb.FileChunk) return fc.loadFromBytes(data) } - func (fc *FilerConf) loadFromBytes(data []byte) (err error) { conf := &filer_pb.FilerConf{} err = proto.UnmarshalText(string(data), conf) diff --git a/weed/filer/storage_option.go b/weed/filer/storage_option.go new file mode 100644 index 000000000..ba9968fcc --- /dev/null +++ b/weed/filer/storage_option.go @@ -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) +} \ No newline at end of file diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index 8f326f5c7..e3f4bb161 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -257,7 +257,15 @@ func (fs *FilerServer) cleanupChunks(existingEntry *filer.Entry, newEntry *filer garbage = append(garbage, coveredChunks...) 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 { // not good, but should be ok 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, err = filer.MaybeManifestize(fs.saveAsChunk(entry.Replication, entry.Collection, "", "", needle.SecondsToTTL(entry.TtlSec), false), entry.Chunks) + so := &filer.StorageOption{ + 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 { // not good, but should be ok glog.V(0).Infof("MaybeManifestize: %v", err) diff --git a/weed/server/filer_server_handlers_write.go b/weed/server/filer_server_handlers_write.go index 267b8752d..e27d9eb26 100644 --- a/weed/server/filer_server_handlers_write.go +++ b/weed/server/filer_server_handlers_write.go @@ -7,6 +7,7 @@ import ( "strings" "time" + "github.com/chrislusf/seaweedfs/weed/filer" "github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/operation" "github.com/chrislusf/seaweedfs/weed/pb/filer_pb" @@ -29,7 +30,7 @@ type FilerPostResult struct { 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() start := time.Now() @@ -37,18 +38,19 @@ func (fs *FilerServer) assignNewFileInfo(replication, collection, dataCenter, ra ar := &operation.VolumeAssignRequest{ Count: 1, - Replication: replication, - Collection: collection, - Ttl: ttlString, - DataCenter: dataCenter, + Replication: so.Replication, + Collection: so.Collection, + Ttl: so.TtlString(), + DataCenter: so.DataCenter, + Rack: so.Rack, } var altRequest *operation.VolumeAssignRequest - if dataCenter != "" || rack != "" { + if so.DataCenter != "" || so.Rack != "" { altRequest = &operation.VolumeAssignRequest{ Count: 1, - Replication: replication, - Collection: collection, - Ttl: ttlString, + Replication: so.Replication, + Collection: so.Collection, + Ttl: so.TtlString(), DataCenter: "", Rack: "", } @@ -62,7 +64,7 @@ func (fs *FilerServer) assignNewFileInfo(replication, collection, dataCenter, ra } fileId = assignResult.Fid urlLocation = "http://" + assignResult.Url + "/" + assignResult.Fid - if fsync { + if so.Fsync { urlLocation += "?fsync=true" } auth = assignResult.Auth @@ -92,7 +94,16 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) { 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) } diff --git a/weed/server/filer_server_handlers_write_autochunk.go b/weed/server/filer_server_handlers_write_autochunk.go index ee0af9aab..178145e69 100644 --- a/weed/server/filer_server_handlers_write_autochunk.go +++ b/weed/server/filer_server_handlers_write_autochunk.go @@ -25,7 +25,7 @@ import ( "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 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, "/") { reply, err = fs.mkdir(ctx, w, r) } 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 { - 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 { 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() if multipartReaderErr != nil { @@ -87,46 +87,46 @@ func (fs *FilerServer) doPostAutoChunk(ctx context.Context, w http.ResponseWrite 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 { 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 { glog.V(0).Infof("manifestize %s: %v", r.RequestURI, replyerr) return } 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 } -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 := "" 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 { 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 { glog.V(0).Infof("manifestize %s: %v", r.RequestURI, replyerr) return } 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 } -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 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), Uid: OS_UID, Gid: OS_GID, - Replication: replication, - Collection: collection, - TtlSec: ttlSec, + Replication: so.Replication, + Collection: so.Collection, + TtlSec: so.TtlSeconds, Mime: contentType, Md5: md5bytes, FileSize: uint64(chunkOffset), @@ -199,7 +199,7 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa 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 md5Hash := md5.New() @@ -211,7 +211,7 @@ func (fs *FilerServer) uploadReaderToChunks(w http.ResponseWriter, r *http.Reque limitedReader := io.LimitReader(partReader, int64(chunkSize)) // 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 { return nil, nil, 0, assignErr } @@ -255,11 +255,11 @@ func (fs *FilerServer) doUpload(urlLocation string, w http.ResponseWriter, r *ht 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) { // 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 { return nil, "", "", assignErr } @@ -270,7 +270,7 @@ func (fs *FilerServer) saveAsChunk(replication string, collection string, dataCe return nil, "", "", uploadErr } - return uploadResult.ToPbFileChunk(fileId, offset), collection, replication, nil + return uploadResult.ToPbFileChunk(fileId, offset), so.Collection, so.Replication, nil } } diff --git a/weed/server/filer_server_handlers_write_cipher.go b/weed/server/filer_server_handlers_write_cipher.go index 720d97027..fa6b0a44f 100644 --- a/weed/server/filer_server_handlers_write_cipher.go +++ b/weed/server/filer_server_handlers_write_cipher.go @@ -16,12 +16,12 @@ import ( ) // 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 == "" { - 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) @@ -65,9 +65,9 @@ func (fs *FilerServer) encrypt(ctx context.Context, w http.ResponseWriter, r *ht Mode: 0660, Uid: OS_UID, Gid: OS_GID, - Replication: replication, - Collection: collection, - TtlSec: ttlSeconds, + Replication: so.Replication, + Collection: so.Collection, + TtlSec: so.TtlSeconds, Mime: pu.MimeType, Md5: util.Base64Md5ToBytes(pu.ContentMd5), },