batch delete on volume servers

3 files changed, 46 insertions, 1 deletions

diff --git a/go/storage/store.go b/go/storage/store.go
index 3e9597e83..6a853122f 100644
--- a/go/storage/store.go
+++ b/go/storage/store.go
@@ -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)
diff --git a/go/weed/weed_server/volume_server.go b/go/weed/weed_server/volume_server.go
index 921f8826f..b522c160f 100644
--- a/go/weed/weed_server/volume_server.go
+++ b/go/weed/weed_server/volume_server.go
@@ -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() {
diff --git a/go/weed/weed_server/volume_server_handlers.go b/go/weed/weed_server/volume_server_handlers.go
index 4f8c7d562..d105cf72e 100644
--- a/go/weed/weed_server/volume_server_handlers.go
+++ b/go/weed/weed_server/volume_server_handlers.go
@@ -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)
+}