Mirror
/
seaweedfs
mirror of https://github.com/seaweedfs/seaweedfs.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

add metrics, refactoring

pull/991/head
Chris Lu 6 years ago
parent
a14bd31f5b
commit
d1cd8f8c5b
4 changed files with 126 additions and 71 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      weed/server/filer_server_handlers_read.go
  2. 4
      weed/server/filer_server_handlers_read_dir.go
  3. 174
      weed/server/filer_server_handlers_write.go
  4. 17
      weed/server/filer_server_handlers_write_autochunk.go

2
weed/server/filer_server_handlers_read.go
View File

@ -61,6 +61,7 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request,
w.Header().Set("Accept-Ranges", "bytes")
if r.Method == "HEAD" {
stats.FilerRequestCounter.WithLabelValues("head").Inc()
w.Header().Set("Content-Length", strconv.FormatInt(int64(filer2.TotalSize(entry.Chunks)), 10))
w.Header().Set("Last-Modified", entry.Attr.Mtime.Format(http.TimeFormat))
setEtag(w, filer2.ETag(entry.Chunks))
@ -88,6 +89,7 @@ func (fs *FilerServer) handleSingleChunk(w http.ResponseWriter, r *http.Request,
}
if fs.option.RedirectOnRead {
stats.FilerRequestCounter.WithLabelValues("redirect").Inc()
http.Redirect(w, r, urlString, http.StatusFound)
return
}

4
weed/server/filer_server_handlers_read_dir.go
View File

@ -9,6 +9,7 @@ import (
"github.com/chrislusf/seaweedfs/weed/filer2"
"github.com/chrislusf/seaweedfs/weed/glog"
ui "github.com/chrislusf/seaweedfs/weed/server/filer_ui"
"github.com/chrislusf/seaweedfs/weed/stats"
)
// listDirectoryHandler lists directories and folers under a directory
@ -16,6 +17,9 @@ import (
// sub directories are listed on the first page, when "lastFileName"
// is empty.
func (fs *FilerServer) listDirectoryHandler(w http.ResponseWriter, r *http.Request) {
stats.FilerRequestCounter.WithLabelValues("list").Inc()
path := r.URL.Path
if strings.HasSuffix(path, "/") && len(path) > 1 {
path = path[:len(path)-1]

174
weed/server/filer_server_handlers_write.go
View File

@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"io/ioutil"
"mime"
@ -38,6 +39,11 @@ type FilerPostResult struct {
}
func (fs *FilerServer) assignNewFileInfo(w http.ResponseWriter, r *http.Request, replication, collection string, dataCenter string) (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 := &operation.VolumeAssignRequest{
Count: 1,
Replication: replication,
@ -116,67 +122,36 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
}
glog.V(4).Infoln("post to", u)
// send request to volume server
request := &http.Request{
Method: r.Method,
URL: u,
Proto: r.Proto,
ProtoMajor: r.ProtoMajor,
ProtoMinor: r.ProtoMinor,
Header: r.Header,
Body: r.Body,
Host: r.Host,
ContentLength: r.ContentLength,
}
if auth != "" {
request.Header.Set("Authorization", "BEARER "+string(auth))
}
resp, do_err := util.Do(request)
if do_err != nil {
glog.Errorf("failing to connect to volume server %s: %v, %+v", r.RequestURI, do_err, r.Method)
writeJsonError(w, r, http.StatusInternalServerError, do_err)
return
}
defer func() {
io.Copy(ioutil.Discard, resp.Body)
resp.Body.Close()
}()
etag := resp.Header.Get("ETag")
resp_body, ra_err := ioutil.ReadAll(resp.Body)
if ra_err != nil {
glog.V(0).Infoln("failing to upload to volume server", r.RequestURI, ra_err.Error())
writeJsonError(w, r, http.StatusInternalServerError, ra_err)
return
}
glog.V(4).Infoln("post result", string(resp_body))
var ret operation.UploadResult
unmarshal_err := json.Unmarshal(resp_body, &ret)
if unmarshal_err != nil {
glog.V(0).Infoln("failing to read upload resonse", r.RequestURI, string(resp_body))
writeJsonError(w, r, http.StatusInternalServerError, unmarshal_err)
return
}
if ret.Error != "" {
glog.V(0).Infoln("failing to post to volume server", r.RequestURI, ret.Error)
writeJsonError(w, r, http.StatusInternalServerError, errors.New(ret.Error))
ret, err := fs.uploadToVolumeServer(r, u, auth, w, fileId)
if err != nil {
return
}
// find correct final path
path := r.URL.Path
if strings.HasSuffix(path, "/") {
if ret.Name != "" {
path += ret.Name
} else {
fs.filer.DeleteFileByFileId(fileId)
glog.V(0).Infoln("Can not to write to folder", path, "without a file name!")
writeJsonError(w, r, http.StatusInternalServerError,
errors.New("Can not to write to folder "+path+" without a file name"))
if err = fs.updateFilerStore(ctx, r, w, replication, collection, ret, fileId); err != nil {
return
}
// send back post result
reply := FilerPostResult{
Name: ret.Name,
Size: ret.Size,
Error: ret.Error,
Fid: fileId,
Url: urlLocation,
}
setEtag(w, ret.ETag)
writeJsonQuiet(w, r, http.StatusCreated, reply)
}
// update metadata in filer store
// update metadata in filer store
func (fs *FilerServer) updateFilerStore(ctx context.Context, r *http.Request, w http.ResponseWriter,
replication string, collection string, ret operation.UploadResult, fileId string) (err error) {
stats.FilerRequestCounter.WithLabelValues("postStoreWrite").Inc()
start := time.Now()
defer func() { stats.FilerRequestHistogram.WithLabelValues("postStoreWrite").Observe(time.Since(start).Seconds()) }()
path := r.URL.Path
existingEntry, err := fs.filer.FindEntry(ctx, filer2.FullPath(path))
crTime := time.Now()
if err == nil && existingEntry != nil {
@ -203,30 +178,95 @@ func (fs *FilerServer) PostHandler(w http.ResponseWriter, r *http.Request) {
FileId: fileId,
Size: uint64(ret.Size),
Mtime: time.Now().UnixNano(),
ETag: etag,
ETag: ret.ETag,
}},
}
if ext := filenamePath.Ext(path); ext != "" {
entry.Attr.Mime = mime.TypeByExtension(ext)
}
// glog.V(4).Infof("saving %s => %+v", path, entry)
if db_err := fs.filer.CreateEntry(ctx, entry); db_err != nil {
if dbErr := fs.filer.CreateEntry(ctx, entry); dbErr != nil {
fs.filer.DeleteChunks(entry.FullPath, entry.Chunks)
glog.V(0).Infof("failing to write %s to filer server : %v", path, db_err)
writeJsonError(w, r, http.StatusInternalServerError, db_err)
glog.V(0).Infof("failing to write %s to filer server : %v", path, dbErr)
writeJsonError(w, r, http.StatusInternalServerError, dbErr)
err = dbErr
return
}
// send back post result
reply := FilerPostResult{
Name: ret.Name,
Size: ret.Size,
Error: ret.Error,
Fid: fileId,
Url: urlLocation,
return nil
}
// send request to volume server
func (fs *FilerServer) uploadToVolumeServer(r *http.Request, u *url.URL, auth security.EncodedJwt, w http.ResponseWriter, fileId string) (ret operation.UploadResult, err error) {
stats.FilerRequestCounter.WithLabelValues("postUpload").Inc()
start := time.Now()
defer func() { stats.FilerRequestHistogram.WithLabelValues("postUpload").Observe(time.Since(start).Seconds()) }()
request := &http.Request{
Method: r.Method,
URL: u,
Proto: r.Proto,
ProtoMajor: r.ProtoMajor,
ProtoMinor: r.ProtoMinor,
Header: r.Header,
Body: r.Body,
Host: r.Host,
ContentLength: r.ContentLength,
}
setEtag(w, etag)
writeJsonQuiet(w, r, http.StatusCreated, reply)
if auth != "" {
request.Header.Set("Authorization", "BEARER "+string(auth))
}
resp, doErr := util.Do(request)
if doErr != nil {
glog.Errorf("failing to connect to volume server %s: %v, %+v", r.RequestURI, doErr, r.Method)
writeJsonError(w, r, http.StatusInternalServerError, doErr)
err = doErr
return
}
defer func() {
io.Copy(ioutil.Discard, resp.Body)
resp.Body.Close()
}()
etag := resp.Header.Get("ETag")
respBody, raErr := ioutil.ReadAll(resp.Body)
if raErr != nil {
glog.V(0).Infoln("failing to upload to volume server", r.RequestURI, raErr.Error())
writeJsonError(w, r, http.StatusInternalServerError, raErr)
err = raErr
return
}
glog.V(4).Infoln("post result", string(respBody))
unmarshalErr := json.Unmarshal(respBody, &ret)
if unmarshalErr != nil {
glog.V(0).Infoln("failing to read upload resonse", r.RequestURI, string(respBody))
writeJsonError(w, r, http.StatusInternalServerError, unmarshalErr)
err = unmarshalErr
return
}
if ret.Error != "" {
err = errors.New(ret.Error)
glog.V(0).Infoln("failing to post to volume server", r.RequestURI, ret.Error)
writeJsonError(w, r, http.StatusInternalServerError, err)
return
}
// find correct final path
path := r.URL.Path
if strings.HasSuffix(path, "/") {
if ret.Name != "" {
path += ret.Name
} else {
err = fmt.Errorf("can not to write to folder %s without a file name", path)
fs.filer.DeleteFileByFileId(fileId)
glog.V(0).Infoln("Can not to write to folder", path, "without a file name!")
writeJsonError(w, r, http.StatusInternalServerError, err)
return
}
}
if etag != "" {
ret.ETag = etag
}
return
}
// curl -X DELETE http://localhost:8888/path/to

17
weed/server/filer_server_handlers_write_autochunk.go
View File

@ -16,6 +16,7 @@ import (
"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/util"
)
@ -68,6 +69,10 @@ func (fs *FilerServer) autoChunk(ctx context.Context, w http.ResponseWriter, r *
func (fs *FilerServer) doAutoChunk(ctx context.Context, w http.ResponseWriter, r *http.Request,
contentLength int64, chunkSize int32, replication string, collection string, dataCenter string) (filerResult *FilerPostResult, replyerr error) {
stats.FilerRequestCounter.WithLabelValues("postAutoChunk").Inc()
start := time.Now()
defer func() { stats.FilerRequestHistogram.WithLabelValues("postAutoChunk").Observe(time.Since(start).Seconds()) }()
multipartReader, multipartReaderErr := r.MultipartReader()
if multipartReaderErr != nil {
return nil, multipartReaderErr
@ -169,11 +174,11 @@ func (fs *FilerServer) doAutoChunk(ctx context.Context, w http.ResponseWriter, r
},
Chunks: fileChunks,
}
if db_err := fs.filer.CreateEntry(ctx, entry); db_err != nil {
if dbErr := fs.filer.CreateEntry(ctx, entry); dbErr != nil {
fs.filer.DeleteChunks(entry.FullPath, entry.Chunks)
replyerr = db_err
filerResult.Error = db_err.Error()
glog.V(0).Infof("failing to write %s to filer server : %v", path, db_err)
replyerr = dbErr
filerResult.Error = dbErr.Error()
glog.V(0).Infof("failing to write %s to filer server : %v", path, dbErr)
return
}
@ -183,6 +188,10 @@ func (fs *FilerServer) doAutoChunk(ctx context.Context, w http.ResponseWriter, r
func (fs *FilerServer) doUpload(urlLocation string, w http.ResponseWriter, r *http.Request,
chunkBuf []byte, fileName string, contentType string, fileId string, auth security.EncodedJwt) (err error) {
stats.FilerRequestCounter.WithLabelValues("postAutoChunkUpload").Inc()
start := time.Now()
defer func() { stats.FilerRequestHistogram.WithLabelValues("postAutoChunkUpload").Observe(time.Since(start).Seconds()) }()
ioReader := ioutil.NopCloser(bytes.NewBuffer(chunkBuf))
uploadResult, uploadError := operation.Upload(urlLocation, fileName, ioReader, false, contentType, nil, auth)
if uploadResult != nil {

Write Preview
Loading…
Cancel
Save