mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2024-12-12 16:49:03 +08:00
162 lines
5.0 KiB
Go
162 lines
5.0 KiB
Go
package weed_server
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"net/http"
|
|
"os"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/chrislusf/seaweedfs/weed/glog"
|
|
"github.com/chrislusf/seaweedfs/weed/operation"
|
|
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
|
|
"github.com/chrislusf/seaweedfs/weed/security"
|
|
"github.com/chrislusf/seaweedfs/weed/stats"
|
|
"github.com/chrislusf/seaweedfs/weed/storage/needle"
|
|
"github.com/chrislusf/seaweedfs/weed/util"
|
|
)
|
|
|
|
var (
|
|
OS_UID = uint32(os.Getuid())
|
|
OS_GID = uint32(os.Getgid())
|
|
|
|
ErrReadOnly = errors.New("read only")
|
|
)
|
|
|
|
type FilerPostResult struct {
|
|
Name string `json:"name,omitempty"`
|
|
Size int64 `json:"size,omitempty"`
|
|
Error string `json:"error,omitempty"`
|
|
}
|
|
|
|
func (fs *FilerServer) assignNewFileInfo(so *operation.StorageOption) (fileId, urlLocation string, auth security.EncodedJwt, err error) {
|
|
|
|
stats.FilerRequestCounter.WithLabelValues("assign").Inc()
|
|
start := time.Now()
|
|
defer func() { stats.FilerRequestHistogram.WithLabelValues("assign").Observe(time.Since(start).Seconds()) }()
|
|
|
|
ar, altRequest := so.ToAssignRequests(1)
|
|
|
|
assignResult, ae := operation.Assign(fs.filer.GetMaster, fs.grpcDialOption, ar, altRequest)
|
|
if ae != nil {
|
|
glog.Errorf("failing to assign a file id: %v", ae)
|
|
err = ae
|
|
return
|
|
}
|
|
fileId = assignResult.Fid
|
|
urlLocation = "http://" + assignResult.Url + "/" + assignResult.Fid
|
|
if so.Fsync {
|
|
urlLocation += "?fsync=true"
|
|
}
|
|
auth = assignResult.Auth
|
|
return
|
|
}
|
|
|
|
func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request, contentLength int64) {
|
|
|
|
ctx := context.Background()
|
|
|
|
query := r.URL.Query()
|
|
so, err := fs.detectStorageOption0(r.RequestURI,
|
|
query.Get("collection"),
|
|
query.Get("replication"),
|
|
query.Get("ttl"),
|
|
query.Get("disk"),
|
|
query.Get("dataCenter"),
|
|
query.Get("rack"),
|
|
)
|
|
if err != nil {
|
|
if err == ErrReadOnly {
|
|
w.WriteHeader(http.StatusInsufficientStorage)
|
|
} else {
|
|
glog.V(1).Infoln("post", r.RequestURI, ":", err.Error())
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
}
|
|
return
|
|
}
|
|
|
|
fs.autoChunk(ctx, w, r, contentLength, so)
|
|
util.CloseRequest(r)
|
|
|
|
}
|
|
|
|
// curl -X DELETE http://localhost:8888/path/to
|
|
// curl -X DELETE http://localhost:8888/path/to?recursive=true
|
|
// curl -X DELETE http://localhost:8888/path/to?recursive=true&ignoreRecursiveError=true
|
|
// curl -X DELETE http://localhost:8888/path/to?recursive=true&skipChunkDeletion=true
|
|
func (fs *FilerServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
|
|
|
|
isRecursive := r.FormValue("recursive") == "true"
|
|
if !isRecursive && fs.option.recursiveDelete {
|
|
if r.FormValue("recursive") != "false" {
|
|
isRecursive = true
|
|
}
|
|
}
|
|
ignoreRecursiveError := r.FormValue("ignoreRecursiveError") == "true"
|
|
skipChunkDeletion := r.FormValue("skipChunkDeletion") == "true"
|
|
|
|
objectPath := r.URL.Path
|
|
if len(r.URL.Path) > 1 && strings.HasSuffix(objectPath, "/") {
|
|
objectPath = objectPath[0 : len(objectPath)-1]
|
|
}
|
|
|
|
err := fs.filer.DeleteEntryMetaAndData(context.Background(), util.FullPath(objectPath), isRecursive, ignoreRecursiveError, !skipChunkDeletion, false, nil)
|
|
if err != nil {
|
|
glog.V(1).Infoln("deleting", objectPath, ":", err.Error())
|
|
httpStatus := http.StatusInternalServerError
|
|
if err == filer_pb.ErrNotFound {
|
|
httpStatus = http.StatusNoContent
|
|
}
|
|
writeJsonError(w, r, httpStatus, err)
|
|
return
|
|
}
|
|
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
|
|
func (fs *FilerServer) detectStorageOption(requestURI, qCollection, qReplication string, ttlSeconds int32, diskType, dataCenter, rack string) (*operation.StorageOption, error) {
|
|
|
|
rule := fs.filer.FilerConf.MatchStorageRule(requestURI)
|
|
|
|
if rule.ReadOnly {
|
|
return nil, ErrReadOnly
|
|
}
|
|
|
|
// required by buckets folder
|
|
bucketDefaultCollection, bucketDefaultReplication, fsync := "", "", false
|
|
if strings.HasPrefix(requestURI, fs.filer.DirBucketsPath+"/") {
|
|
bucketDefaultCollection = fs.filer.DetectBucket(util.FullPath(requestURI))
|
|
bucketDefaultReplication, fsync = fs.filer.ReadBucketOption(bucketDefaultCollection)
|
|
}
|
|
|
|
if ttlSeconds == 0 {
|
|
ttl, err := needle.ReadTTL(rule.GetTtl())
|
|
if err != nil {
|
|
glog.Errorf("fail to parse %s ttl setting %s: %v", rule.LocationPrefix, rule.Ttl, err)
|
|
}
|
|
ttlSeconds = int32(ttl.Minutes()) * 60
|
|
}
|
|
|
|
return &operation.StorageOption{
|
|
Replication: util.Nvl(qReplication, rule.Replication, bucketDefaultReplication, fs.option.DefaultReplication),
|
|
Collection: util.Nvl(qCollection, rule.Collection, bucketDefaultCollection, fs.option.Collection),
|
|
DataCenter: util.Nvl(dataCenter, fs.option.DataCenter),
|
|
Rack: util.Nvl(rack, fs.option.Rack),
|
|
TtlSeconds: ttlSeconds,
|
|
DiskType: util.Nvl(diskType, rule.DiskType),
|
|
Fsync: fsync || rule.Fsync,
|
|
VolumeGrowthCount: rule.VolumeGrowthCount,
|
|
}, nil
|
|
}
|
|
|
|
func (fs *FilerServer) detectStorageOption0(requestURI, qCollection, qReplication string, qTtl string, diskType string, dataCenter, rack 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)
|
|
}
|