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

batch delete on volume servers

Browse source
This commit is contained in:
Chris Lu 2014-04-14 01:00:09 -07:00
parent 460923f094
commit 56a3d30e75
3 changed files with 46 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
go/storage/store.go
View file

@ -352,7 +352,7 @@ func (s *Store) Read(i VolumeId, n *Needle) (int, error) {
if v := s.findVolume(i); v != nil {
return v.read(n)
}
return 0, fmt.Errorf("Volume %s not found!", i)
return 0, fmt.Errorf("Volume %v not found!", i)
}
func (s *Store) GetVolume(i VolumeId) *Volume {
return s.findVolume(i)

1
go/weed/weed_server/volume_server.go
View file

@ -41,6 +41,7 @@ func NewVolumeServer(r *http.ServeMux, ip string, port int, publicUrl string, fo
r.HandleFunc("/stats/counter", secure(vs.whiteList, statsCounterHandler))
r.HandleFunc("/stats/memory", secure(vs.whiteList, statsMemoryHandler))
r.HandleFunc("/stats/disk", secure(vs.whiteList, vs.statsDiskHandler))
r.HandleFunc("/delete", secure(vs.whiteList, vs.batchDeleteHandler))
r.HandleFunc("/", vs.storeHandler)
go func() {

44
go/weed/weed_server/volume_server_handlers.go
View file

@ -195,3 +195,47 @@ func (vs *VolumeServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
m["size"] = uint32(count)
writeJsonQuiet(w, r, m)
}
type DeleteResult struct {
Fid string `json:"fid"`
Size int `json:"size"`
Error string `json:"error,omitempty"`
}
//Experts only: takes multiple fid parameters. This function does not propagate deletes to replicas.
func (vs *VolumeServer) batchDeleteHandler(w http.ResponseWriter, r *http.Request) {
r.ParseForm()
var ret []DeleteResult
for _, fid := range r.Form["fid"] {
n := new(storage.Needle)
commaIndex := strings.Index(fid, ",")
if commaIndex <= 0 {
ret = append(ret, DeleteResult{Fid: fid, Error: "Wrong fid format."})
continue
}
vid := fid[:commaIndex]
volumeId, _ := storage.NewVolumeId(vid)
id_cookie := fid[commaIndex+1:]
n.ParsePath(id_cookie)
glog.V(4).Infoln("batch deleting", n)
cookie := n.Cookie
if _, err := vs.store.Read(volumeId, n); err != nil {
ret = append(ret, DeleteResult{Fid: fid, Error: err.Error()})
continue
}
if n.Cookie != cookie {
ret = append(ret, DeleteResult{Fid: fid, 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, DeleteResult{Fid: fid, Error: err.Error()})
} else {
ret = append(ret, DeleteResult{Fid: fid, Size: int(size)})
}
}
w.WriteHeader(http.StatusAccepted)
writeJsonQuiet(w, r, ret)
}