seaweedfs/weed/server/volume_server_handlers.go

176 lines
5.2 KiB
Go
Raw Normal View History

package weed_server
import (
2022-05-20 10:18:20 +00:00
"fmt"
"net/http"
"strconv"
"strings"
"sync/atomic"
2022-05-20 10:18:20 +00:00
"time"
2020-10-31 23:31:39 +00:00
"github.com/chrislusf/seaweedfs/weed/util"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/stats"
)
/*
If volume server is started with a separated public port, the public port will
be more "secure".
Public port currently only supports reads.
Later writes on public port can have one of the 3
security settings:
1. not secured
2. secured by white list
3. secured by JWT(Json Web Token)
*/
func (vs *VolumeServer) privateStoreHandler(w http.ResponseWriter, r *http.Request) {
2020-09-20 23:01:56 +00:00
w.Header().Set("Server", "SeaweedFS Volume "+util.VERSION)
2020-10-31 23:31:39 +00:00
if r.Header.Get("Origin") != "" {
w.Header().Set("Access-Control-Allow-Origin", "*")
w.Header().Set("Access-Control-Allow-Credentials", "true")
}
switch r.Method {
2018-07-22 17:27:10 +00:00
case "GET", "HEAD":
stats.ReadRequest()
vs.inFlightDownloadDataLimitCond.L.Lock()
for vs.concurrentDownloadLimit != 0 && atomic.LoadInt64(&vs.inFlightDownloadDataSize) > vs.concurrentDownloadLimit {
select {
case <-r.Context().Done():
glog.V(4).Infof("request cancelled from %s: %v", r.RemoteAddr, r.Context().Err())
return
default:
glog.V(4).Infof("wait because inflight download data %d > %d", vs.inFlightDownloadDataSize, vs.concurrentDownloadLimit)
vs.inFlightDownloadDataLimitCond.Wait()
}
}
vs.inFlightDownloadDataLimitCond.L.Unlock()
2014-05-27 04:15:05 +00:00
vs.GetOrHeadHandler(w, r)
case "DELETE":
stats.DeleteRequest()
vs.guard.WhiteList(vs.DeleteHandler)(w, r)
2018-07-22 17:27:10 +00:00
case "PUT", "POST":
contentLength := getContentLength(r)
// exclude the replication from the concurrentUploadLimitMB
if r.URL.Query().Get("type") != "replicate" && vs.concurrentUploadLimit != 0 {
startTime := time.Now()
vs.inFlightUploadDataLimitCond.L.Lock()
for vs.inFlightUploadDataSize > vs.concurrentUploadLimit {
//wait timeout check
2022-05-20 10:18:20 +00:00
if startTime.Add(vs.inflightUploadDataTimeout).Before(time.Now()) {
vs.inFlightUploadDataLimitCond.L.Unlock()
err := fmt.Errorf("reject because inflight upload data %d > %d, and wait timeout", vs.inFlightUploadDataSize, vs.concurrentUploadLimit)
2022-05-20 10:18:20 +00:00
glog.V(1).Infof("too many requests: %v", err)
writeJsonError(w, r, http.StatusTooManyRequests, err)
return
}
glog.V(4).Infof("wait because inflight upload data %d > %d", vs.inFlightUploadDataSize, vs.concurrentUploadLimit)
vs.inFlightUploadDataLimitCond.Wait()
}
vs.inFlightUploadDataLimitCond.L.Unlock()
}
atomic.AddInt64(&vs.inFlightUploadDataSize, contentLength)
defer func() {
atomic.AddInt64(&vs.inFlightUploadDataSize, -contentLength)
if vs.concurrentUploadLimit != 0 {
vs.inFlightUploadDataLimitCond.Signal()
}
}()
// processs uploads
stats.WriteRequest()
vs.guard.WhiteList(vs.PostHandler)(w, r)
2020-10-31 23:31:39 +00:00
case "OPTIONS":
stats.ReadRequest()
2020-10-31 23:44:03 +00:00
w.Header().Add("Access-Control-Allow-Methods", "PUT, POST, GET, DELETE, OPTIONS")
w.Header().Add("Access-Control-Allow-Headers", "*")
}
}
func getContentLength(r *http.Request) int64 {
contentLength := r.Header.Get("Content-Length")
if contentLength != "" {
length, err := strconv.ParseInt(contentLength, 10, 64)
if err != nil {
return 0
}
return length
}
return 0
}
func (vs *VolumeServer) publicReadOnlyHandler(w http.ResponseWriter, r *http.Request) {
2020-09-20 23:01:56 +00:00
w.Header().Set("Server", "SeaweedFS Volume "+util.VERSION)
2020-10-31 23:31:39 +00:00
if r.Header.Get("Origin") != "" {
w.Header().Set("Access-Control-Allow-Origin", "*")
w.Header().Set("Access-Control-Allow-Credentials", "true")
}
switch r.Method {
case "GET", "HEAD":
stats.ReadRequest()
vs.inFlightDownloadDataLimitCond.L.Lock()
for vs.concurrentDownloadLimit != 0 && atomic.LoadInt64(&vs.inFlightDownloadDataSize) > vs.concurrentDownloadLimit {
glog.V(4).Infof("wait because inflight download data %d > %d", vs.inFlightDownloadDataSize, vs.concurrentDownloadLimit)
vs.inFlightDownloadDataLimitCond.Wait()
}
vs.inFlightDownloadDataLimitCond.L.Unlock()
vs.GetOrHeadHandler(w, r)
2020-10-31 23:31:39 +00:00
case "OPTIONS":
stats.ReadRequest()
2020-10-31 23:44:03 +00:00
w.Header().Add("Access-Control-Allow-Methods", "GET, OPTIONS")
w.Header().Add("Access-Control-Allow-Headers", "*")
2014-04-14 08:00:09 +00:00
}
}
2019-06-06 07:29:02 +00:00
func (vs *VolumeServer) maybeCheckJwtAuthorization(r *http.Request, vid, fid string, isWrite bool) bool {
2019-06-06 07:29:02 +00:00
var signingKey security.SigningKey
if isWrite {
if len(vs.guard.SigningKey) == 0 {
return true
} else {
signingKey = vs.guard.SigningKey
}
2019-06-11 04:33:32 +00:00
} else {
2019-06-06 07:29:02 +00:00
if len(vs.guard.ReadSigningKey) == 0 {
return true
} else {
signingKey = vs.guard.ReadSigningKey
}
}
tokenStr := security.GetJwt(r)
if tokenStr == "" {
glog.V(1).Infof("missing jwt from %s", r.RemoteAddr)
return false
}
token, err := security.DecodeJwt(signingKey, tokenStr, &security.SeaweedFileIdClaims{})
if err != nil {
glog.V(1).Infof("jwt verification error from %s: %v", r.RemoteAddr, err)
return false
}
if !token.Valid {
glog.V(1).Infof("jwt invalid from %s: %v", r.RemoteAddr, tokenStr)
return false
}
if sc, ok := token.Claims.(*security.SeaweedFileIdClaims); ok {
if sepIndex := strings.LastIndex(fid, "_"); sepIndex > 0 {
fid = fid[:sepIndex]
}
return sc.Fid == vid+","+fid
}
glog.V(1).Infof("unexpected jwt from %s: %v", r.RemoteAddr, tokenStr)
return false
}