gsh-digital-services/seaweedFS
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #2530 from banjiaojuhao/filer-upload-file-to-node

Browse Source 
filer server: allow upload file to specific dataNode
This commit is contained in:
Chris Lu
2021-12-22 12:49:15 -08:00
committed by GitHub
parent b7cd52636b 08336be92e
commit 4e73705533
8 changed files with 345 additions and 326 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
weed/server/filer_grpc_server.go
View File

@@ -218,6 +218,7 @@ func (fs *FilerServer) cleanupChunks(fullpath string, existingEntry *filer.Entry
newEntry.Attributes.DiskType,
"",
"",
"",
) // ignore readonly error for capacity needed to manifestize
chunks, err = filer.MaybeManifestize(fs.saveAsChunk(so), chunks)
if err != nil {
@@ -259,7 +260,7 @@ func (fs *FilerServer) AppendToEntry(ctx context.Context, req *filer_pb.AppendTo
}
entry.Chunks = append(entry.Chunks, req.Chunks...)
so, err := fs.detectStorageOption(string(fullpath), entry.Collection, entry.Replication, entry.TtlSec, entry.DiskType, "", "")
so, err := fs.detectStorageOption(string(fullpath), entry.Collection, entry.Replication, entry.TtlSec, entry.DiskType, "", "", "")
if err != nil {
glog.Warningf("detectStorageOption: %v", err)
return &filer_pb.AppendToEntryResponse{}, err
@@ -289,7 +290,7 @@ func (fs *FilerServer) DeleteEntry(ctx context.Context, req *filer_pb.DeleteEntr
func (fs *FilerServer) AssignVolume(ctx context.Context, req *filer_pb.AssignVolumeRequest) (resp *filer_pb.AssignVolumeResponse, err error) {
so, err := fs.detectStorageOption(req.Path, req.Collection, req.Replication, req.TtlSec, req.DiskType, req.DataCenter, req.Rack)
so, err := fs.detectStorageOption(req.Path, req.Collection, req.Replication, req.TtlSec, req.DiskType, req.DataCenter, req.Rack, req.DataNode)
if err != nil {
glog.V(3).Infof("AssignVolume: %v", err)
return &filer_pb.AssignVolumeResponse{Error: fmt.Sprintf("assign volume: %v", err)}, nil

2
weed/server/filer_grpc_server_remote.go
View File

@@ -64,7 +64,7 @@ func (fs *FilerServer) CacheRemoteObjectToLocalCluster(ctx context.Context, req
}
// detect storage option
so, err := fs.detectStorageOption(req.Directory, "", "", 0, "", "", "")
so, err := fs.detectStorageOption(req.Directory, "", "", 0, "", "", "", "")
if err != nil {
return resp, err
}

1
weed/server/filer_server.go
View File

@@ -56,6 +56,7 @@ type FilerOption struct {
DirListingLimit int
DataCenter string
Rack string
DataNode string
DefaultLevelDbDir string
DisableHttp bool
Host pb.ServerAddress

8
weed/server/filer_server_handlers_write.go
View File

@@ -65,6 +65,7 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request, conte
query.Get("disk"),
query.Get("dataCenter"),
query.Get("rack"),
query.Get("dataNode"),
)
if err != nil {
if err == ErrReadOnly {
@@ -115,7 +116,7 @@ func (fs *FilerServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusNoContent)
}
func (fs *FilerServer) detectStorageOption(requestURI, qCollection, qReplication string, ttlSeconds int32, diskType, dataCenter, rack string) (*operation.StorageOption, error) {
func (fs *FilerServer) detectStorageOption(requestURI, qCollection, qReplication string, ttlSeconds int32, diskType, dataCenter, rack, dataNode string) (*operation.StorageOption, error) {
rule := fs.filer.FilerConf.MatchStorageRule(requestURI)
@@ -143,6 +144,7 @@ func (fs *FilerServer) detectStorageOption(requestURI, qCollection, qReplication
Collection: util.Nvl(qCollection, rule.Collection, bucketDefaultCollection, fs.option.Collection),
DataCenter: util.Nvl(dataCenter, fs.option.DataCenter),
Rack: util.Nvl(rack, fs.option.Rack),
DataNode: util.Nvl(dataNode, fs.option.DataNode),
TtlSeconds: ttlSeconds,
DiskType: util.Nvl(diskType, rule.DiskType),
Fsync: fsync || rule.Fsync,
@@ -150,12 +152,12 @@ func (fs *FilerServer) detectStorageOption(requestURI, qCollection, qReplication
}, nil
}
func (fs *FilerServer) detectStorageOption0(requestURI, qCollection, qReplication string, qTtl string, diskType string, dataCenter, rack string) (*operation.StorageOption, error) {
func (fs *FilerServer) detectStorageOption0(requestURI, qCollection, qReplication string, qTtl string, diskType string, dataCenter, rack, dataNode string) (*operation.StorageOption, error) {
ttl, err := needle.ReadTTL(qTtl)
if err != nil {
glog.Errorf("fail to parse ttl %s: %v", qTtl, err)
}
return fs.detectStorageOption(requestURI, qCollection, qReplication, int32(ttl.Minutes())*60, diskType, dataCenter, rack)
return fs.detectStorageOption(requestURI, qCollection, qReplication, int32(ttl.Minutes())*60, diskType, dataCenter, rack, dataNode)
}