gitlab-com-seaweedfs/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-01-19 02:48:24 +00:00
Code Issues Projects Releases Wiki Activity

more check in http_util.Delete

Browse source 
add status code in `DeleteResult` struct
operation.DeleteFiles maybe unsafe, so `ChunkManifest.DeleteChunks` manually delete each chunks
This commit is contained in:
tnextday 2015-12-03 16:27:02 +08:00
parent a9a336fdff
commit daac5de1ba
4 changed files with 70 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
go/operation/chunked_file.go
View file

@ -70,19 +70,11 @@ func (cm *ChunkManifest) GetData() ([]byte, error) {
}
func (cm *ChunkManifest) DeleteChunks(master string) error {
fileIds := make([]string, 0, len(cm.Chunks))
for _, ci := range cm.Chunks {
fileIds = append(fileIds, ci.Fid)
}
results, e := DeleteFiles(master, fileIds)
if e != nil {
return e
}
deleteError := 0
for _, ret := range results.Results {
if ret.Error != "" {
for _, ci := range cm.Chunks {
if e := DeleteFile(master, ci.Fid, ""); e != nil {
deleteError++
glog.V(0).Infoln("delete error:", ret.Error, ret.Fid)
glog.V(0).Infoln("delete error:", e, ci.Fid)
}
}
if deleteError > 0 {

16
go/operation/delete_content.go
View file

@ -7,14 +7,17 @@ import (
"strings"
"sync"
"net/http"
"github.com/chrislusf/seaweedfs/go/security"
"github.com/chrislusf/seaweedfs/go/util"
)
type DeleteResult struct {
Fid string `json:"fid"`
Size int `json:"size"`
Error string `json:"error,omitempty"`
Fid string `json:"fid"`
Size int `json:"size"`
Status int `json:"status"`
Error string `json:"error,omitempty"`
}
func DeleteFile(master string, fileId string, jwt security.EncodedJwt) error {
@ -45,7 +48,11 @@ func DeleteFiles(master string, fileIds []string) (*DeleteFilesResult, error) {
for _, fileId := range fileIds {
vid, _, err := ParseFileId(fileId)
if err != nil {
ret.Results = append(ret.Results, DeleteResult{Fid: vid, Error: err.Error()})
ret.Results = append(ret.Results, DeleteResult{
Fid: vid,
Status: http.StatusBadRequest,
Error: err.Error()},
)
continue
}
if _, ok := vid_to_fileIds[vid]; !ok {
@ -76,6 +83,7 @@ func DeleteFiles(master string, fileIds []string) (*DeleteFilesResult, error) {
}
var wg sync.WaitGroup
for server, fidList := range server_to_fileIds {
wg.Add(1)
go func(server string, fidList []string) {

18
go/util/http_util.go
View file

@ -9,7 +9,10 @@ import (
"net/url"
"strings"
"encoding/json"
"github.com/chrislusf/seaweedfs/go/security"
"github.com/syndtr/goleveldb/leveldb/errors"
)
var (
@ -79,10 +82,21 @@ func Delete(url string, jwt security.EncodedJwt) error {
return e
}
defer resp.Body.Close()
if _, err := ioutil.ReadAll(resp.Body); err != nil {
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return err
}
return nil
switch resp.StatusCode {
case http.StatusNotFound, http.StatusAccepted, http.StatusOK:
return nil
}
m := make(map[string]interface{})
if e := json.Unmarshal(body, m); e == nil {
if s, ok := m["error"].(string); ok {
return errors.New(s)
}
}
return errors.New(string(body))
}
func GetBufferStream(url string, values url.Values, allocatedBytes []byte, eachBuffer func([]byte)) error {

47
go/weed/weed_server/volume_server_handlers_write.go
View file

@ -63,19 +63,20 @@ func (vs *VolumeServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
if n.Cookie != cookie {
glog.V(0).Infoln("delete", r.URL.Path, "with unmaching cookie from ", r.RemoteAddr, "agent", r.UserAgent())
writeJsonError(w, r, http.StatusBadRequest, errors.New("File Random Cookie does not match."))
return
}
count := int64(n.Size)
if n.IsChunkedManifest(){
if n.IsChunkedManifest() {
chunkManifest, e := operation.LoadChunkManifest(n.Data, n.IsGzipped())
if e != nil {
writeJsonError(w, r, http.StatusInternalServerError, errors.New("Load chunks manifest error: " + e.Error()))
writeJsonError(w, r, http.StatusInternalServerError, errors.New("Load chunks manifest error: "+e.Error()))
return
}
if e := chunkManifest.DeleteChunks(vs.GetMasterNode()); e != nil {
writeJsonError(w, r, http.StatusInternalServerError, errors.New("Delete chunks error: " + e.Error()))
writeJsonError(w, r, http.StatusInternalServerError, errors.New("Delete chunks error: "+e.Error()))
return
}
count = chunkManifest.Size
@ -100,7 +101,10 @@ func (vs *VolumeServer) batchDeleteHandler(w http.ResponseWriter, r *http.Reques
for _, fid := range r.Form["fid"] {
vid, id_cookie, err := operation.ParseFileId(fid)
if err != nil {
ret = append(ret, operation.DeleteResult{Fid: fid, Error: err.Error()})
ret = append(ret, operation.DeleteResult{
Fid: fid,
Status: http.StatusBadRequest,
Error: err.Error()})
continue
}
n := new(storage.Needle)
@ -109,18 +113,45 @@ func (vs *VolumeServer) batchDeleteHandler(w http.ResponseWriter, r *http.Reques
glog.V(4).Infoln("batch deleting", n)
cookie := n.Cookie
if _, err := vs.store.ReadVolumeNeedle(volumeId, n); err != nil {
ret = append(ret, operation.DeleteResult{Fid: fid, Error: err.Error()})
ret = append(ret, operation.DeleteResult{
Fid: fid,
Status: http.StatusNotFound,
Error: err.Error(),
})
continue
}
if n.IsChunkedManifest() {
//Don't allow delete manifest in batch delete mode
ret = append(ret, operation.DeleteResult{
Fid: fid,
Status: http.StatusNotAcceptable,
Error: "ChunkManifest: not allow.",
})
continue
}
if n.Cookie != cookie {
ret = append(ret, operation.DeleteResult{Fid: fid, Error: "File Random Cookie does not match."})
ret = append(ret, operation.DeleteResult{
Fid: fid,
Status: http.StatusBadRequest,
Error: "File Random Cookie does not match.",
})
glog.V(0).Infoln("deleting", fid, "with unmaching cookie from ", r.RemoteAddr, "agent", r.UserAgent())
return
}
if size, err := vs.store.Delete(volumeId, n); err != nil {
ret = append(ret, operation.DeleteResult{Fid: fid, Error: err.Error()})
ret = append(ret, operation.DeleteResult{
Fid: fid,
Status: http.StatusInternalServerError,
Error: err.Error()},
)
} else {
ret = append(ret, operation.DeleteResult{Fid: fid, Size: int(size)})
ret = append(ret, operation.DeleteResult{
Fid: fid,
Status: http.StatusAccepted,
Size: int(size)},
)
}
}